[hornetq-commits] JBoss hornetq SVN: r8947 - in trunk: src/main/org/hornetq/api/core and 51 other directories.

Author: timfox Date: 2010-03-23 11:11:11 -0400 (Tue, 23 Mar 2010) New Revision: 8947 Added: trunk/src/main/org/hornetq/utils/ConcurrentHQDeque.java trunk/src/main/org/hornetq/utils/PriorityLinkedList.java trunk/src/main/org/hornetq/utils/PriorityLinkedListImpl.java trunk/src/main/org/hornetq/utils/concurrent/HQIterator.java trunk/src/main/org/hornetq/utils/concurrent/HornetQConcurrentLinkedQueue.java trunk/tests/src/org/hornetq/tests/integration/client/NIOvsOIOTest.java trunk/tests/src/org/hornetq/tests/unit/core/list/impl/PriorityLinkedListTestBase.java trunk/tests/src/org/hornetq/tests/unit/util/concurrent/HornetQConcurrentLinkedQueueTest.java Removed: trunk/src/main/org/hornetq/core/list/PriorityHeadInsertableQueue.java trunk/src/main/org/hornetq/core/list/PriorityLinkedList.java trunk/src/main/org/hornetq/core/list/impl/ trunk/src/main/org/hornetq/core/server/impl/ServerProducerCreditManager.java trunk/src/main/org/hornetq/core/server/impl/ServerProducerCreditManagerImpl.java trunk/src/main/org/hornetq/integration/transports/netty/ChannelPipelineSupport.java trunk/tests/src/org/hornetq/tests/integration/client/CoreSelectorTest.java trunk/tests/src/org/hornetq/tests/integration/remoting/AardvarkProtocolTest.java trunk/tests/src/org/hornetq/tests/unit/core/list/impl/PriorityLinkedListTest.java trunk/tests/src/org/hornetq/tests/unit/core/remoting/impl/netty/ChannelBufferWrapperTest.java Modified: trunk/pom.xml trunk/src/main/org/hornetq/api/core/SimpleString.java trunk/src/main/org/hornetq/api/core/client/HornetQClient.java trunk/src/main/org/hornetq/api/core/management/QueueControl.java trunk/src/main/org/hornetq/api/jms/management/JMSQueueControl.java trunk/src/main/org/hornetq/core/client/impl/ClientConsumerImpl.java trunk/src/main/org/hornetq/core/client/impl/ClientProducerCreditManager.java trunk/src/main/org/hornetq/core/client/impl/ClientProducerCreditManagerImpl.java trunk/src/main/org/hornetq/core/client/impl/ClientProducerCredits.java trunk/src/main/org/hornetq/core/client/impl/ClientProducerCreditsImpl.java trunk/src/main/org/hornetq/core/client/impl/ClientProducerImpl.java trunk/src/main/org/hornetq/core/client/impl/ClientSessionImpl.java trunk/src/main/org/hornetq/core/client/impl/ClientSessionInternal.java trunk/src/main/org/hornetq/core/client/impl/ClientSessionPacketHandler.java trunk/src/main/org/hornetq/core/client/impl/DelegatingSession.java trunk/src/main/org/hornetq/core/client/impl/FailoverManagerImpl.java trunk/src/main/org/hornetq/core/client/impl/LargeMessageBufferImpl.java trunk/src/main/org/hornetq/core/config/impl/ConfigurationImpl.java trunk/src/main/org/hornetq/core/management/impl/QueueControlImpl.java trunk/src/main/org/hornetq/core/messagecounter/MessageCounter.java trunk/src/main/org/hornetq/core/paging/PagingManager.java trunk/src/main/org/hornetq/core/paging/PagingStore.java trunk/src/main/org/hornetq/core/paging/impl/PagingManagerImpl.java trunk/src/main/org/hornetq/core/paging/impl/PagingStoreImpl.java trunk/src/main/org/hornetq/core/paging/impl/TestSupportPageStore.java trunk/src/main/org/hornetq/core/persistence/impl/journal/LargeServerMessageImpl.java trunk/src/main/org/hornetq/core/postoffice/PostOffice.java trunk/src/main/org/hornetq/core/postoffice/impl/BindingsImpl.java trunk/src/main/org/hornetq/core/postoffice/impl/PostOfficeImpl.java trunk/src/main/org/hornetq/core/protocol/aardvark/impl/AardvarkProtocolManager.java trunk/src/main/org/hornetq/core/protocol/core/Channel.java trunk/src/main/org/hornetq/core/protocol/core/CoreRemotingConnection.java trunk/src/main/org/hornetq/core/protocol/core/impl/ChannelImpl.java trunk/src/main/org/hornetq/core/protocol/core/impl/CoreProtocolManager.java trunk/src/main/org/hornetq/core/protocol/core/impl/CoreSessionCallback.java trunk/src/main/org/hornetq/core/protocol/core/impl/HornetQPacketHandler.java trunk/src/main/org/hornetq/core/protocol/core/impl/RemotingConnectionImpl.java trunk/src/main/org/hornetq/core/protocol/core/impl/wireformat/SessionProducerCreditsMessage.java trunk/src/main/org/hornetq/core/protocol/core/impl/wireformat/SessionReceiveLargeMessage.java trunk/src/main/org/hornetq/core/protocol/stomp/StompSession.java trunk/src/main/org/hornetq/core/protocol/stomp/StompUtils.java trunk/src/main/org/hornetq/core/remoting/impl/invm/InVMConnection.java trunk/src/main/org/hornetq/core/remoting/server/impl/RemotingServiceImpl.java trunk/src/main/org/hornetq/core/replication/impl/ReplicationEndpointImpl.java trunk/src/main/org/hornetq/core/server/Queue.java trunk/src/main/org/hornetq/core/server/ServerMessage.java trunk/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java trunk/src/main/org/hornetq/core/server/impl/QueueImpl.java trunk/src/main/org/hornetq/core/server/impl/ServerConsumerImpl.java trunk/src/main/org/hornetq/core/server/impl/ServerInfo.java trunk/src/main/org/hornetq/core/server/impl/ServerMessageImpl.java trunk/src/main/org/hornetq/core/server/impl/ServerSessionImpl.java trunk/src/main/org/hornetq/integration/transports/netty/HornetQChannelHandler.java trunk/src/main/org/hornetq/integration/transports/netty/NettyAcceptor.java trunk/src/main/org/hornetq/integration/transports/netty/NettyConnection.java trunk/src/main/org/hornetq/integration/transports/netty/NettyConnector.java trunk/src/main/org/hornetq/integration/transports/netty/TransportConstants.java trunk/src/main/org/hornetq/jms/client/HornetQMessage.java trunk/src/main/org/hornetq/jms/client/HornetQMessageProducer.java trunk/src/main/org/hornetq/jms/management/impl/JMSQueueControlImpl.java trunk/src/main/org/hornetq/spi/core/protocol/SessionCallback.java trunk/src/main/org/hornetq/spi/core/remoting/Connection.java trunk/src/main/org/hornetq/utils/HornetQThreadFactory.java trunk/src/main/org/hornetq/utils/TypedProperties.java trunk/tests/jms-tests/src/org/hornetq/jms/tests/message/JMSMessageIDHeaderTest.java trunk/tests/jms-tests/src/org/hornetq/jms/tests/message/MessageHeaderTest.java trunk/tests/jms-tests/src/org/hornetq/jms/tests/selector/SelectorTest.java trunk/tests/src/org/hornetq/tests/integration/client/AutogroupIdTest.java trunk/tests/src/org/hornetq/tests/integration/client/ConsumerTest.java trunk/tests/src/org/hornetq/tests/integration/client/ExpiryAddressTest.java trunk/tests/src/org/hornetq/tests/integration/client/LargeMessageTest.java trunk/tests/src/org/hornetq/tests/integration/client/MessageGroupingConnectionFactoryTest.java trunk/tests/src/org/hornetq/tests/integration/client/MessageGroupingTest.java trunk/tests/src/org/hornetq/tests/integration/client/MessageHandlerTest.java trunk/tests/src/org/hornetq/tests/integration/client/MessagePriorityTest.java trunk/tests/src/org/hornetq/tests/integration/client/OrderTest.java trunk/tests/src/org/hornetq/tests/integration/client/PagingTest.java trunk/tests/src/org/hornetq/tests/integration/client/ProducerFlowControlTest.java trunk/tests/src/org/hornetq/tests/integration/client/ReceiveTest.java trunk/tests/src/org/hornetq/tests/integration/cluster/failover/FailoverTest.java trunk/tests/src/org/hornetq/tests/integration/cluster/failover/LargeMessageFailoverTest.java trunk/tests/src/org/hornetq/tests/integration/divert/PersistentDivertTest.java trunk/tests/src/org/hornetq/tests/integration/jms/client/TextMessageTest.java trunk/tests/src/org/hornetq/tests/integration/jms/server/management/JMSQueueControlTest.java trunk/tests/src/org/hornetq/tests/integration/jms/server/management/JMSQueueControlUsingJMSTest.java trunk/tests/src/org/hornetq/tests/integration/journal/JournalPerfTuneTest.java trunk/tests/src/org/hornetq/tests/integration/largemessage/LargeMessageTestBase.java trunk/tests/src/org/hornetq/tests/integration/management/QueueControlUsingCoreTest.java trunk/tests/src/org/hornetq/tests/integration/paging/PageCrashTest.java trunk/tests/src/org/hornetq/tests/integration/paging/PagingSendTest.java trunk/tests/src/org/hornetq/tests/integration/security/NettySecurityClientTest.java trunk/tests/src/org/hornetq/tests/integration/security/SimpleClient.java trunk/tests/src/org/hornetq/tests/integration/xa/BasicXaRecoveryTest.java trunk/tests/src/org/hornetq/tests/unit/core/postoffice/impl/BindingsImplTest.java trunk/tests/src/org/hornetq/tests/unit/core/postoffice/impl/FakeQueue.java trunk/tests/src/org/hornetq/tests/unit/core/remoting/impl/netty/NettyConnectionTest.java trunk/tests/src/org/hornetq/tests/unit/core/server/impl/QueueImplTest.java trunk/tests/src/org/hornetq/tests/unit/core/server/impl/fakes/FakeQueueFactory.java trunk/tests/src/org/hornetq/tests/util/JMSTestBase.java Log: rewrote delivery algorithm, simplified producer flow control, various other optimisations, fixes and shit like that Modified: trunk/pom.xml =================================================================== --- trunk/pom.xml 2010-03-22 23:43:26 UTC (rev 8946) +++ trunk/pom.xml 2010-03-23 15:11:11 UTC (rev 8947) @@ -222,7 +222,7 @@ <dependency> <groupId>org.jboss.netty</groupId> <artifactId>netty</artifactId> - <version>3.1.5.GA</version> + <version>3.2.0.BETA1</version> </dependency> <!--needed to compile the logging jar--> <dependency> Modified: trunk/src/main/org/hornetq/api/core/SimpleString.java =================================================================== --- trunk/src/main/org/hornetq/api/core/SimpleString.java 2010-03-22 23:43:26 UTC (rev 8946) +++ trunk/src/main/org/hornetq/api/core/SimpleString.java 2010-03-23 15:11:11 UTC (rev 8947) @@ -211,6 +211,11 @@ @Override public boolean equals(final Object other) { + if (this == other) + { + return true; + } + if (other instanceof SimpleString) { SimpleString s = (SimpleString)other; Modified: trunk/src/main/org/hornetq/api/core/client/HornetQClient.java =================================================================== --- trunk/src/main/org/hornetq/api/core/client/HornetQClient.java 2010-03-22 23:43:26 UTC (rev 8946) +++ trunk/src/main/org/hornetq/api/core/client/HornetQClient.java 2010-03-23 15:11:11 UTC (rev 8947) @@ -48,7 +48,7 @@ public static final int DEFAULT_CONFIRMATION_WINDOW_SIZE = -1; - public static final int DEFAULT_PRODUCER_WINDOW_SIZE = 1024 * 1024; + public static final int DEFAULT_PRODUCER_WINDOW_SIZE = 64 * 1024; public static final int DEFAULT_PRODUCER_MAX_RATE = -1; Modified: trunk/src/main/org/hornetq/api/core/management/QueueControl.java =================================================================== --- trunk/src/main/org/hornetq/api/core/management/QueueControl.java 2010-03-22 23:43:26 UTC (rev 8946) +++ trunk/src/main/org/hornetq/api/core/management/QueueControl.java 2010-03-23 15:11:11 UTC (rev 8947) @@ -80,7 +80,7 @@ /** * Returns the number of messages added to this queue since it was created. */ - int getMessagesAdded(); + long getMessagesAdded(); /** * Returns the expiry address associated to this queue. Modified: trunk/src/main/org/hornetq/api/jms/management/JMSQueueControl.java =================================================================== --- trunk/src/main/org/hornetq/api/jms/management/JMSQueueControl.java 2010-03-22 23:43:26 UTC (rev 8946) +++ trunk/src/main/org/hornetq/api/jms/management/JMSQueueControl.java 2010-03-23 15:11:11 UTC (rev 8947) @@ -52,7 +52,7 @@ /** * Returns the number of messages added to this queue since it was created. */ - int getMessagesAdded(); + long getMessagesAdded(); /** * Returns the number of scheduled messages in this queue. Modified: trunk/src/main/org/hornetq/core/client/impl/ClientConsumerImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/client/impl/ClientConsumerImpl.java 2010-03-22 23:43:26 UTC (rev 8946) +++ trunk/src/main/org/hornetq/core/client/impl/ClientConsumerImpl.java 2010-03-23 15:11:11 UTC (rev 8947) @@ -22,8 +22,6 @@ import org.hornetq.api.core.SimpleString; import org.hornetq.api.core.client.ClientMessage; import org.hornetq.api.core.client.MessageHandler; -import org.hornetq.core.list.PriorityLinkedList; -import org.hornetq.core.list.impl.PriorityLinkedListImpl; import org.hornetq.core.logging.Logger; import org.hornetq.core.protocol.core.Channel; import org.hornetq.core.protocol.core.impl.wireformat.SessionConsumerCloseMessage; @@ -32,7 +30,10 @@ import org.hornetq.core.protocol.core.impl.wireformat.SessionReceiveContinuationMessage; import org.hornetq.core.protocol.core.impl.wireformat.SessionReceiveLargeMessage; import org.hornetq.utils.Future; +import org.hornetq.utils.PriorityLinkedList; +import org.hornetq.utils.PriorityLinkedListImpl; import org.hornetq.utils.TokenBucketLimiter; +import org.hornetq.utils.concurrent.HQIterator; /** * @author <a href="mailto:tim.fox@jboss.com">Tim Fox</a> @@ -78,7 +79,7 @@ private final int ackBatchSize; - private final PriorityLinkedList<ClientMessageInternal> buffer = new PriorityLinkedListImpl<ClientMessageInternal>(ClientConsumerImpl.NUM_PRIORITIES); + private final PriorityLinkedList<ClientMessageInternal> buffer = new PriorityLinkedListImpl<ClientMessageInternal>(false, ClientConsumerImpl.NUM_PRIORITIES); private final Runner runner = new Runner(); @@ -108,7 +109,7 @@ private volatile int ackBytes; - private volatile ClientMessage lastAckedMessage; + private volatile ClientMessageInternal lastAckedMessage; private boolean stopped = false; @@ -454,7 +455,7 @@ { return browseOnly; } - + public synchronized void handleMessage(final ClientMessageInternal message) throws Exception { if (closing) @@ -462,11 +463,11 @@ // This is ok - we just ignore the message return; } - + ClientMessageInternal messageToHandle = message; messageToHandle.onReceipt(this); - + // Add it to the buffer buffer.addLast(messageToHandle, messageToHandle.getPriority()); @@ -539,7 +540,11 @@ { // Need to send credits for the messages in the buffer - for (ClientMessageInternal message : buffer) + HQIterator<ClientMessageInternal> iter = buffer.iterator(); + + ClientMessageInternal message; + + while ((message = iter.next()) != null) { flowControlBeforeConsumption(message); } @@ -564,15 +569,17 @@ public void acknowledge(final ClientMessage message) throws HornetQException { + ClientMessageInternal cmi = (ClientMessageInternal)message; + ackBytes += message.getEncodeSize(); if (ackBytes >= ackBatchSize) { - doAck(message); + doAck(cmi); } else { - lastAckedMessage = message; + lastAckedMessage = cmi; } } @@ -863,7 +870,7 @@ buffer.clear(); } - private void doAck(final ClientMessage message) throws HornetQException + private void doAck(final ClientMessageInternal message) throws HornetQException { ackBytes = 0; Modified: trunk/src/main/org/hornetq/core/client/impl/ClientProducerCreditManager.java =================================================================== --- trunk/src/main/org/hornetq/core/client/impl/ClientProducerCreditManager.java 2010-03-22 23:43:26 UTC (rev 8946) +++ trunk/src/main/org/hornetq/core/client/impl/ClientProducerCreditManager.java 2010-03-23 15:11:11 UTC (rev 8947) @@ -28,7 +28,7 @@ void returnCredits(SimpleString address); - void receiveCredits(SimpleString address, int credits, int offset); + void receiveCredits(SimpleString address, int credits); void reset(); Modified: trunk/src/main/org/hornetq/core/client/impl/ClientProducerCreditManagerImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/client/impl/ClientProducerCreditManagerImpl.java 2010-03-22 23:43:26 UTC (rev 8946) +++ trunk/src/main/org/hornetq/core/client/impl/ClientProducerCreditManagerImpl.java 2010-03-23 15:11:11 UTC (rev 8947) @@ -85,13 +85,13 @@ } } - public synchronized void receiveCredits(final SimpleString address, final int credits, final int offset) + public synchronized void receiveCredits(final SimpleString address, final int credits) { ClientProducerCredits cr = producerCredits.get(address); if (cr != null) { - cr.receiveCredits(credits, offset); + cr.receiveCredits(credits); } } Modified: trunk/src/main/org/hornetq/core/client/impl/ClientProducerCredits.java =================================================================== --- trunk/src/main/org/hornetq/core/client/impl/ClientProducerCredits.java 2010-03-22 23:43:26 UTC (rev 8946) +++ trunk/src/main/org/hornetq/core/client/impl/ClientProducerCredits.java 2010-03-23 15:11:11 UTC (rev 8947) @@ -24,7 +24,7 @@ { void acquireCredits(int credits) throws InterruptedException; - void receiveCredits(int credits, int offset); + void receiveCredits(int credits); void reset(); Modified: trunk/src/main/org/hornetq/core/client/impl/ClientProducerCreditsImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/client/impl/ClientProducerCreditsImpl.java 2010-03-22 23:43:26 UTC (rev 8946) +++ trunk/src/main/org/hornetq/core/client/impl/ClientProducerCreditsImpl.java 2010-03-23 15:11:11 UTC (rev 8947) @@ -62,20 +62,18 @@ public void acquireCredits(final int credits) throws InterruptedException { - // credits += offset; - checkCredits(credits); semaphore.acquire(credits); } - public void receiveCredits(final int credits, final int offset) + public void receiveCredits(final int credits) { synchronized (this) { arriving -= credits; } - + semaphore.release(credits); } @@ -109,9 +107,7 @@ public synchronized void releaseOutstanding() { - int permits = semaphore.drainPermits(); - - session.sendProducerCreditsMessage(permits, address); + semaphore.drainPermits(); } private void checkCredits(final int credits) Modified: trunk/src/main/org/hornetq/core/client/impl/ClientProducerImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/client/impl/ClientProducerImpl.java 2010-03-22 23:43:26 UTC (rev 8946) +++ trunk/src/main/org/hornetq/core/client/impl/ClientProducerImpl.java 2010-03-23 15:11:11 UTC (rev 8947) @@ -217,14 +217,18 @@ if (address != null) { - msgI.setAddress(address); + //session.setAddress(msg, address); + + msg.setAddress(address); // Anonymous theCredits = session.getCredits(address, true); } else { - msgI.setAddress(this.address); + //session.setAddress(msg, this.address); + + msg.setAddress(this.address); theCredits = credits; } Modified: trunk/src/main/org/hornetq/core/client/impl/ClientSessionImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/client/impl/ClientSessionImpl.java 2010-03-22 23:43:26 UTC (rev 8946) +++ trunk/src/main/org/hornetq/core/client/impl/ClientSessionImpl.java 2010-03-23 15:11:11 UTC (rev 8947) @@ -26,6 +26,7 @@ import org.hornetq.api.core.HornetQBuffer; import org.hornetq.api.core.HornetQException; +import org.hornetq.api.core.Message; import org.hornetq.api.core.SimpleString; import org.hornetq.api.core.client.ClientConsumer; import org.hornetq.api.core.client.ClientMessage; @@ -707,7 +708,7 @@ } else { - channel.send(message); + channel.sendBatched(message); } } @@ -1016,6 +1017,29 @@ // not having any credits to send } } + + private volatile SimpleString defaultAddress; + + public void setAddress(final Message message, final SimpleString address) + { + if (defaultAddress == null) + { + defaultAddress = address; + + message.setAddress(address); + } + else + { + if (!address.equals(defaultAddress)) + { + message.setAddress(address); + } + else + { + message.setAddress(null); + } + } + } private void sendPacketWithoutLock(final Packet packet) { @@ -1058,9 +1082,9 @@ producerCreditManager.returnCredits(address); } - public void handleReceiveProducerCredits(final SimpleString address, final int credits, final int offset) + public void handleReceiveProducerCredits(final SimpleString address, final int credits) { - producerCreditManager.receiveCredits(address, credits, offset); + producerCreditManager.receiveCredits(address, credits); } public ClientProducerCreditManager getProducerCreditManager() Modified: trunk/src/main/org/hornetq/core/client/impl/ClientSessionInternal.java =================================================================== --- trunk/src/main/org/hornetq/core/client/impl/ClientSessionInternal.java 2010-03-22 23:43:26 UTC (rev 8946) +++ trunk/src/main/org/hornetq/core/client/impl/ClientSessionInternal.java 2010-03-23 15:11:11 UTC (rev 8947) @@ -14,6 +14,7 @@ package org.hornetq.core.client.impl; import org.hornetq.api.core.HornetQException; +import org.hornetq.api.core.Message; import org.hornetq.api.core.SimpleString; import org.hornetq.api.core.client.ClientSession; import org.hornetq.core.protocol.core.CoreRemotingConnection; @@ -75,7 +76,9 @@ void returnCredits(SimpleString address); - void handleReceiveProducerCredits(SimpleString address, int credits, int offset); + void handleReceiveProducerCredits(SimpleString address, int credits); ClientProducerCreditManager getProducerCreditManager(); + + void setAddress(Message message, SimpleString address); } Modified: trunk/src/main/org/hornetq/core/client/impl/ClientSessionPacketHandler.java =================================================================== --- trunk/src/main/org/hornetq/core/client/impl/ClientSessionPacketHandler.java 2010-03-22 23:43:26 UTC (rev 8946) +++ trunk/src/main/org/hornetq/core/client/impl/ClientSessionPacketHandler.java 2010-03-23 15:11:11 UTC (rev 8947) @@ -88,8 +88,7 @@ SessionProducerCreditsMessage message = (SessionProducerCreditsMessage)packet; clientSession.handleReceiveProducerCredits(message.getAddress(), - message.getCredits(), - message.getOffset()); + message.getCredits()); break; } Modified: trunk/src/main/org/hornetq/core/client/impl/DelegatingSession.java =================================================================== --- trunk/src/main/org/hornetq/core/client/impl/DelegatingSession.java 2010-03-22 23:43:26 UTC (rev 8946) +++ trunk/src/main/org/hornetq/core/client/impl/DelegatingSession.java 2010-03-23 15:11:11 UTC (rev 8947) @@ -20,6 +20,7 @@ import javax.transaction.xa.Xid; import org.hornetq.api.core.HornetQException; +import org.hornetq.api.core.Message; import org.hornetq.api.core.SimpleString; import org.hornetq.api.core.client.ClientConsumer; import org.hornetq.api.core.client.ClientMessage; @@ -530,13 +531,18 @@ session.returnCredits(address); } - public void handleReceiveProducerCredits(final SimpleString address, final int credits, final int offset) + public void handleReceiveProducerCredits(final SimpleString address, final int credits) { - session.handleReceiveProducerCredits(address, credits, offset); + session.handleReceiveProducerCredits(address, credits); } public ClientProducerCreditManager getProducerCreditManager() { return session.getProducerCreditManager(); } + + public void setAddress(Message message, SimpleString address) + { + session.setAddress(message, address); + } } Modified: trunk/src/main/org/hornetq/core/client/impl/FailoverManagerImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/client/impl/FailoverManagerImpl.java 2010-03-22 23:43:26 UTC (rev 8946) +++ trunk/src/main/org/hornetq/core/client/impl/FailoverManagerImpl.java 2010-03-23 15:11:11 UTC (rev 8947) @@ -1186,6 +1186,8 @@ Channel channel0 = connection.getChannel(0, -1); channel0.send(ping); + + connection.flush(); } public synchronized void cancel() Modified: trunk/src/main/org/hornetq/core/client/impl/LargeMessageBufferImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/client/impl/LargeMessageBufferImpl.java 2010-03-22 23:43:26 UTC (rev 8946) +++ trunk/src/main/org/hornetq/core/client/impl/LargeMessageBufferImpl.java 2010-03-23 15:11:11 UTC (rev 8947) @@ -1280,6 +1280,7 @@ { throw new IllegalAccessError("Can't read the messageBody after setting outputStream"); } + if (index >= totalSize) { throw new IndexOutOfBoundsException(); Modified: trunk/src/main/org/hornetq/core/config/impl/ConfigurationImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/config/impl/ConfigurationImpl.java 2010-03-22 23:43:26 UTC (rev 8946) +++ trunk/src/main/org/hornetq/core/config/impl/ConfigurationImpl.java 2010-03-23 15:11:11 UTC (rev 8947) @@ -62,7 +62,7 @@ public static final int DEFAULT_SCHEDULED_THREAD_POOL_MAX_SIZE = 5; - public static final int DEFAULT_THREAD_POOL_MAX_SIZE = -1; + public static final int DEFAULT_THREAD_POOL_MAX_SIZE = 30; public static final long DEFAULT_SECURITY_INVALIDATION_INTERVAL = 10000; Deleted: trunk/src/main/org/hornetq/core/list/PriorityHeadInsertableQueue.java =================================================================== --- trunk/src/main/org/hornetq/core/list/PriorityHeadInsertableQueue.java 2010-03-22 23:43:26 UTC (rev 8946) +++ trunk/src/main/org/hornetq/core/list/PriorityHeadInsertableQueue.java 2010-03-23 15:11:11 UTC (rev 8947) @@ -1,47 +0,0 @@ -/* - * Copyright 2009 Red Hat, Inc. - * Red Hat licenses this file to you under the Apache License, version - * 2.0 (the "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * http://www.apache.org/licenses/LICENSE-2.0 - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or - * implied. See the License for the specific language governing - * permissions and limitations under the License. - */ - -package org.hornetq.core.list; - -import java.util.Iterator; -import java.util.List; - -/** - * A type of linked queue which maintains items according to a priority - * and allows adding and removing of elements at both ends, and peeking - * - * @author <a href="mailto:tim.fox@jboss.com>Tim Fox</a> - * @version <tt>$Revision: 1174 $</tt> - * - * $Id: PrioritizedDeque.java 1174 2006-08-02 14:14:32Z timfox $ - */ -public interface PriorityHeadInsertableQueue<T> extends Iterable<T> -{ - void offerFirst(T t, int priority); - - void offerLast(T t, int priority); - - T poll(); - - T peek(); - - List<T> getAll(); - - void clear(); - - int size(); - - Iterator<T> iterator(); - - boolean isEmpty(); -} Deleted: trunk/src/main/org/hornetq/core/list/PriorityLinkedList.java =================================================================== --- trunk/src/main/org/hornetq/core/list/PriorityLinkedList.java 2010-03-22 23:43:26 UTC (rev 8946) +++ trunk/src/main/org/hornetq/core/list/PriorityLinkedList.java 2010-03-23 15:11:11 UTC (rev 8947) @@ -1,48 +0,0 @@ -/* - * Copyright 2009 Red Hat, Inc. - * Red Hat licenses this file to you under the Apache License, version - * 2.0 (the "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * http://www.apache.org/licenses/LICENSE-2.0 - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or - * implied. See the License for the specific language governing - * permissions and limitations under the License. - */ - -package org.hornetq.core.list; - -import java.util.Iterator; -import java.util.List; - -/** - * A type of linked list which maintains items according to a priority - * and allows adding and removing of elements at both ends, and peeking - * - * @author <a href="mailto:tim.fox@jboss.com>Tim Fox</a> - * @version <tt>$Revision: 1174 $</tt> - * - * $Id: PrioritizedDeque.java 1174 2006-08-02 14:14:32Z timfox $ - */ -public interface PriorityLinkedList<T> extends Iterable<T> -{ - void addFirst(T t, int priority); - - void addLast(T t, int priority); - - T removeFirst(); - - T peekFirst(); - - List<T> getAll(); - - void clear(); - - int size(); - - Iterator<T> iterator(); - - boolean isEmpty(); - -} Modified: trunk/src/main/org/hornetq/core/management/impl/QueueControlImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/management/impl/QueueControlImpl.java 2010-03-22 23:43:26 UTC (rev 8946) +++ trunk/src/main/org/hornetq/core/management/impl/QueueControlImpl.java 2010-03-23 15:11:11 UTC (rev 8947) @@ -198,7 +198,7 @@ } } - public int getMessagesAdded() + public long getMessagesAdded() { clearIO(); try Modified: trunk/src/main/org/hornetq/core/messagecounter/MessageCounter.java =================================================================== --- trunk/src/main/org/hornetq/core/messagecounter/MessageCounter.java 2010-03-22 23:43:26 UTC (rev 8946) +++ trunk/src/main/org/hornetq/core/messagecounter/MessageCounter.java 2010-03-23 15:11:11 UTC (rev 8947) @@ -77,7 +77,7 @@ private final List<DayCounter> dayCounters; - private int lastMessagesAdded; + private long lastMessagesAdded; // Static -------------------------------------------------------- @@ -123,9 +123,9 @@ */ public synchronized void onTimer() { - int latestMessagesAdded = destQueue.getMessagesAdded(); + long latestMessagesAdded = destQueue.getMessagesAdded(); - int newMessagesAdded = latestMessagesAdded - lastMessagesAdded; + long newMessagesAdded = latestMessagesAdded - lastMessagesAdded; countTotal += newMessagesAdded; @@ -350,7 +350,7 @@ * * @param newMessages number of new messages to add to the latest day counter */ - private void updateHistory(final int newMessages) + private void updateHistory(final long newMessages) { // check history activation if (dayCounters.isEmpty()) @@ -492,7 +492,7 @@ * * @param newMessages number of new messages since the counter was last updated. */ - void updateDayCounter(final int newMessages) + void updateDayCounter(final long newMessages) { // get the current hour of the day GregorianCalendar cal = new GregorianCalendar(); Modified: trunk/src/main/org/hornetq/core/paging/PagingManager.java =================================================================== --- trunk/src/main/org/hornetq/core/paging/PagingManager.java 2010-03-22 23:43:26 UTC (rev 8946) +++ trunk/src/main/org/hornetq/core/paging/PagingManager.java 2010-03-23 15:11:11 UTC (rev 8947) @@ -68,14 +68,7 @@ */ void removeTransaction(long transactionID); - long getTotalMemory(); - /** - * @param size - */ - long addSize(long size); - - /** * Reload previously created PagingStores into memory * @throws Exception */ Modified: trunk/src/main/org/hornetq/core/paging/PagingStore.java =================================================================== --- trunk/src/main/org/hornetq/core/paging/PagingStore.java 2010-03-22 23:43:26 UTC (rev 8946) +++ trunk/src/main/org/hornetq/core/paging/PagingStore.java 2010-03-23 15:11:11 UTC (rev 8947) @@ -17,7 +17,6 @@ import org.hornetq.core.server.HornetQComponent; import org.hornetq.core.server.MessageReference; import org.hornetq.core.server.ServerMessage; -import org.hornetq.core.server.impl.ServerProducerCreditManager; import org.hornetq.core.settings.impl.AddressFullMessagePolicy; /** @@ -44,6 +43,8 @@ long getPageSizeBytes(); long getAddressSize(); + + long getMaxSize(); boolean isPaging(); @@ -61,13 +62,7 @@ */ boolean startDepaging(); - void addSize(ServerMessage message, boolean add); - - void addSize(MessageReference reference, boolean add); - - int getAvailableProducerCredits(int desired); - - void returnProducerCredits(int credits); - - ServerProducerCreditManager getProducerCreditManager(); + void addSize(int size); + + void executeRunnableWhenMemoryAvailable(Runnable runnable); } Modified: trunk/src/main/org/hornetq/core/paging/impl/PagingManagerImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/paging/impl/PagingManagerImpl.java 2010-03-22 23:43:26 UTC (rev 8946) +++ trunk/src/main/org/hornetq/core/paging/impl/PagingManagerImpl.java 2010-03-23 15:11:11 UTC (rev 8947) @@ -17,7 +17,6 @@ import java.util.Set; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; -import java.util.concurrent.atomic.AtomicLong; import org.hornetq.api.core.SimpleString; import org.hornetq.core.logging.Logger; @@ -45,8 +44,6 @@ private volatile boolean started = false; - private final AtomicLong totalMemoryBytes = new AtomicLong(0); - private final ConcurrentMap<SimpleString, PagingStore> stores = new ConcurrentHashMap<SimpleString, PagingStore>(); private final HierarchicalRepository<AddressSettings> addressSettingsRepository; @@ -198,8 +195,6 @@ } pagingStoreFactory.stop(); - - totalMemoryBytes.set(0); } public void resumeDepages() @@ -224,22 +219,6 @@ } } - /* (non-Javadoc) - * @see org.hornetq.core.paging.PagingManager#getGlobalSize() - */ - public long getTotalMemory() - { - return totalMemoryBytes.get(); - } - - /* (non-Javadoc) - * @see org.hornetq.core.paging.PagingManager#addGlobalSize(long) - */ - public long addSize(final long size) - { - return totalMemoryBytes.addAndGet(size); - } - // Package protected --------------------------------------------- // Protected ----------------------------------------------------- Modified: trunk/src/main/org/hornetq/core/paging/impl/PagingStoreImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/paging/impl/PagingStoreImpl.java 2010-03-22 23:43:26 UTC (rev 8946) +++ trunk/src/main/org/hornetq/core/paging/impl/PagingStoreImpl.java 2010-03-23 15:11:11 UTC (rev 8947) @@ -17,6 +17,8 @@ import java.text.DecimalFormat; import java.util.HashSet; import java.util.List; +import java.util.Queue; +import java.util.concurrent.ConcurrentLinkedQueue; import java.util.concurrent.CountDownLatch; import java.util.concurrent.Executor; import java.util.concurrent.TimeUnit; @@ -41,11 +43,7 @@ import org.hornetq.core.persistence.StorageManager; import org.hornetq.core.postoffice.PostOffice; import org.hornetq.core.server.LargeServerMessage; -import org.hornetq.core.server.MessageReference; import org.hornetq.core.server.ServerMessage; -import org.hornetq.core.server.impl.MessageReferenceImpl; -import org.hornetq.core.server.impl.ServerProducerCreditManager; -import org.hornetq.core.server.impl.ServerProducerCreditManagerImpl; import org.hornetq.core.settings.impl.AddressFullMessagePolicy; import org.hornetq.core.settings.impl.AddressSettings; import org.hornetq.core.transaction.Transaction; @@ -118,14 +116,8 @@ * */ private final ReadWriteLock currentPageLock = new ReentrantReadWriteLock(); - private final ServerProducerCreditManager creditManager; - - private boolean exceededAvailableCredits; - private volatile boolean running = false; - private final AtomicLong availableProducerCredits = new AtomicLong(0); - protected final boolean syncNonTransactional; // Static -------------------------------------------------------- @@ -190,10 +182,6 @@ this.storeFactory = storeFactory; - availableProducerCredits.set(maxSize); - - creditManager = new ServerProducerCreditManagerImpl(this); - this.syncNonTransactional = syncNonTransactional; } @@ -211,6 +199,11 @@ return sizeInBytes.get(); } + public long getMaxSize() + { + return maxSize; + } + public AddressFullMessagePolicy getAddressFullMessagePolicy() { return addressFullMessagePolicy; @@ -252,75 +245,6 @@ return storeName; } - public boolean isExceededAvailableCredits() - { - return exceededAvailableCredits; - } - - public synchronized int getAvailableProducerCredits(final int credits) - { - if (maxSize != -1 && addressFullMessagePolicy == AddressFullMessagePolicy.BLOCK) - { - long avail = availableProducerCredits.get(); - - if (avail > 0) - { - long take = Math.min(avail, credits); - - availableProducerCredits.addAndGet(-take); - - return (int)take; - } - - return 0; - } - else - { - return credits; - } - } - - public void returnProducerCredits(final int credits) - { - checkReleaseProducerFlowControlCredits(-credits); - } - - public void addSize(final ServerMessage message, final boolean add) - { - long size = message.getMemoryEstimate(); - - if (add) - { - checkReleaseProducerFlowControlCredits(size); - - addSize(size); - } - else - { - checkReleaseProducerFlowControlCredits(-size); - - addSize(-size); - } - } - - public void addSize(final MessageReference reference, final boolean add) - { - long size = MessageReferenceImpl.getMemoryEstimate(); - - if (add) - { - checkReleaseProducerFlowControlCredits(size); - - addSize(size); - } - else - { - checkReleaseProducerFlowControlCredits(-size); - - addSize(-size); - } - } - public boolean page(final ServerMessage message, final long transactionID, final boolean duplicateDetection) throws Exception { // The sync on transactions is done on commit only @@ -579,11 +503,6 @@ return new PageImpl(storeName, storageManager, fileFactory, file, page); } - public ServerProducerCreditManager getProducerCreditManager() - { - return creditManager; - } - // TestSupportPageStore ------------------------------------------ public void forceAnotherPage() throws Exception @@ -708,44 +627,68 @@ } - private synchronized void checkReleaseProducerFlowControlCredits(final long size) + private Queue<Runnable> onMemoryFreedRunnables = new ConcurrentLinkedQueue<Runnable>(); + + private class MemoryFreedRunnablesExecutor implements Runnable { - if (addressFullMessagePolicy == AddressFullMessagePolicy.BLOCK && maxSize != -1) + public void run() { - long avail = availableProducerCredits.addAndGet(-size); + Runnable runnable; - if (avail > 0) + while ((runnable = onMemoryFreedRunnables.poll()) != null) { - int used = creditManager.creditsReleased((int)avail); + runnable.run(); + } + } + } - long num = availableProducerCredits.addAndGet(-used); + private final Runnable memoryFreedRunnablesExecutor = new MemoryFreedRunnablesExecutor(); - if (num < 0) + private final Object runnableLock = new Object(); + + public void executeRunnableWhenMemoryAvailable(final Runnable runnable) + { + if (addressFullMessagePolicy == AddressFullMessagePolicy.BLOCK && maxSize != -1) + { + synchronized (runnableLock) + { + if (sizeInBytes.get() > maxSize) { - PagingStoreImpl.log.warn("Available credits has gone negative"); + onMemoryFreedRunnables.add(runnable); - exceededAvailableCredits = true; + return; } } } + runnable.run(); } - private void addSize(final long size) + public void addSize(final int size) { - if (addressFullMessagePolicy != AddressFullMessagePolicy.PAGE) + if (addressFullMessagePolicy == AddressFullMessagePolicy.BLOCK) { - addAddressSize(size); + if (maxSize != -1) + { + synchronized (runnableLock) + { + long newSize = sizeInBytes.addAndGet(size); - pagingManager.addSize(size); + if (newSize <= maxSize) + { + if (!onMemoryFreedRunnables.isEmpty()) + { + executor.execute(memoryFreedRunnablesExecutor); + } + } + } + } return; } - else + else if (addressFullMessagePolicy == AddressFullMessagePolicy.PAGE) { - pagingManager.addSize(size); + final long addressSize = sizeInBytes.addAndGet(size); - final long addressSize = addAddressSize(size); - if (size > 0) { if (maxSize > 0 && addressSize > maxSize) @@ -779,6 +722,11 @@ return; } + else if (addressFullMessagePolicy == AddressFullMessagePolicy.DROP) + { + sizeInBytes.addAndGet(size); + } + } private boolean page(final ServerMessage message, @@ -1063,11 +1011,6 @@ return maxSize > 0 && getAddressSize() + nextPageSize > maxSize; } - private long addAddressSize(final long delta) - { - return sizeInBytes.addAndGet(delta); - } - /** * startDepaging and clearDepage needs to be atomic. * We can't use writeLock to this operation as writeLock would still be used by another thread, and still being a valid usage @@ -1080,8 +1023,6 @@ if (PagingStoreImpl.isTrace) { PagingStoreImpl.trace("Clear Depage on Address = " + getStoreName() + - " PagingManager size " + - pagingManager.getTotalMemory() + " addressSize = " + getAddressSize() + " addressMax " + Modified: trunk/src/main/org/hornetq/core/paging/impl/TestSupportPageStore.java =================================================================== --- trunk/src/main/org/hornetq/core/paging/impl/TestSupportPageStore.java 2010-03-22 23:43:26 UTC (rev 8946) +++ trunk/src/main/org/hornetq/core/paging/impl/TestSupportPageStore.java 2010-03-23 15:11:11 UTC (rev 8947) @@ -36,8 +36,6 @@ void forceAnotherPage() throws Exception; - boolean isExceededAvailableCredits(); - /** @return true if paging was started, or false if paging was already started before this call */ boolean startPaging() throws Exception; Modified: trunk/src/main/org/hornetq/core/persistence/impl/journal/LargeServerMessageImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/persistence/impl/journal/LargeServerMessageImpl.java 2010-03-22 23:43:26 UTC (rev 8946) +++ trunk/src/main/org/hornetq/core/persistence/impl/journal/LargeServerMessageImpl.java 2010-03-23 15:11:11 UTC (rev 8947) @@ -22,7 +22,6 @@ import org.hornetq.core.logging.Logger; import org.hornetq.core.message.BodyEncoder; import org.hornetq.core.server.LargeServerMessage; -import org.hornetq.core.server.MessageReference; import org.hornetq.core.server.ServerMessage; import org.hornetq.core.server.impl.ServerMessageImpl; import org.hornetq.utils.DataConstants; @@ -194,9 +193,9 @@ } @Override - public synchronized int decrementRefCount(final MessageReference reference) throws Exception + public synchronized int decrementRefCount() throws Exception { - int currentRefCount = super.decrementRefCount(reference); + int currentRefCount = super.decrementRefCount(); // We use <= as this could be used by load. // because of a failure, no references were loaded, so we have 0... and we still need to delete the associated @@ -316,9 +315,8 @@ file = storageManager.createFileForLargeMessage(getMessageID(), durable); file.open(); - + bodySize = file.size(); - } } catch (Exception e) Modified: trunk/src/main/org/hornetq/core/postoffice/PostOffice.java =================================================================== --- trunk/src/main/org/hornetq/core/postoffice/PostOffice.java 2010-03-22 23:43:26 UTC (rev 8946) +++ trunk/src/main/org/hornetq/core/postoffice/PostOffice.java 2010-03-23 15:11:11 UTC (rev 8947) @@ -34,9 +34,6 @@ * * A Queue instance can only be bound against a single address in the post office. * - * The PostOffice also maintains a set of "allowable addresses". These are the addresses that it is legal to - * route to. - * * @author <a href="mailto:tim.fox@jboss.com">Tim Fox</a> * */ @@ -51,7 +48,7 @@ Binding getBinding(SimpleString uniqueName); Bindings getMatchingBindings(SimpleString address); - + void route(ServerMessage message) throws Exception; void route(ServerMessage message, Transaction tx) throws Exception; Modified: trunk/src/main/org/hornetq/core/postoffice/impl/BindingsImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/postoffice/impl/BindingsImpl.java 2010-03-22 23:43:26 UTC (rev 8946) +++ trunk/src/main/org/hornetq/core/postoffice/impl/BindingsImpl.java 2010-03-23 15:11:11 UTC (rev 8947) @@ -281,9 +281,9 @@ Integer ipos = routingNamePositions.get(routingName); int pos = ipos != null ? ipos : 0; - + int length = bindings.size(); - + int startPos = pos; Binding theBinding = null; @@ -371,7 +371,10 @@ break; } } - routingNamePositions.put(routingName, pos); + if (pos != startPos) + { + routingNamePositions.put(routingName, pos); + } return theBinding; } Modified: trunk/src/main/org/hornetq/core/postoffice/impl/PostOfficeImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/postoffice/impl/PostOfficeImpl.java 2010-03-22 23:43:26 UTC (rev 8946) +++ trunk/src/main/org/hornetq/core/postoffice/impl/PostOfficeImpl.java 2010-03-23 15:11:11 UTC (rev 8947) @@ -64,7 +64,7 @@ import org.hornetq.core.transaction.TransactionPropertyIndexes; import org.hornetq.core.transaction.Transaction.State; import org.hornetq.core.transaction.impl.TransactionImpl; -import org.hornetq.utils.ExecutorFactory; +import org.hornetq.utils.ConcurrentHashSet; import org.hornetq.utils.TypedProperties; import org.hornetq.utils.UUIDGenerator; @@ -538,7 +538,7 @@ { this.route(message, new RoutingContextImpl(tx)); } - + public void route(final ServerMessage message, final RoutingContext context) throws Exception { // Sanity check @@ -685,7 +685,7 @@ message.incrementDurableRefCount(); - message.incrementRefCount(reference); + message.incrementRefCount(); if (tx == null) { @@ -869,7 +869,7 @@ reference.setScheduledDeliveryTime(scheduledDeliveryTime); } - message.incrementRefCount(reference); + message.incrementRefCount(); } Iterator<Queue> iter = context.getDurableQueues().iterator(); @@ -929,7 +929,7 @@ } } - message.incrementRefCount(reference); + message.incrementRefCount(); } if (tx != null) @@ -1268,7 +1268,7 @@ message.decrementDurableRefCount(); } - message.decrementRefCount(ref); + message.decrementRefCount(); } } } Modified: trunk/src/main/org/hornetq/core/protocol/aardvark/impl/AardvarkProtocolManager.java =================================================================== --- trunk/src/main/org/hornetq/core/protocol/aardvark/impl/AardvarkProtocolManager.java 2010-03-22 23:43:26 UTC (rev 8946) +++ trunk/src/main/org/hornetq/core/protocol/aardvark/impl/AardvarkProtocolManager.java 2010-03-23 15:11:11 UTC (rev 8947) @@ -142,7 +142,7 @@ return -1; } - public void sendProducerCreditsMessage(int credits, SimpleString address, int offset) + public void sendProducerCreditsMessage(int credits, SimpleString address) { } Modified: trunk/src/main/org/hornetq/core/protocol/core/Channel.java =================================================================== --- trunk/src/main/org/hornetq/core/protocol/core/Channel.java 2010-03-22 23:43:26 UTC (rev 8946) +++ trunk/src/main/org/hornetq/core/protocol/core/Channel.java 2010-03-23 15:11:11 UTC (rev 8947) @@ -43,6 +43,13 @@ * @param packet the packet to send */ void send(Packet packet); + + /** + * sends a packet on this channel using batching algorithm if appropriate + * + * @param packet the packet to send + */ + void sendBatched(Packet packet); /** * sends a packet on this channel and then blocks until it has been written to the connection. Modified: trunk/src/main/org/hornetq/core/protocol/core/CoreRemotingConnection.java =================================================================== --- trunk/src/main/org/hornetq/core/protocol/core/CoreRemotingConnection.java 2010-03-22 23:43:26 UTC (rev 8946) +++ trunk/src/main/org/hornetq/core/protocol/core/CoreRemotingConnection.java 2010-03-23 15:11:11 UTC (rev 8947) @@ -86,4 +86,10 @@ * @return the lock */ Object getTransferLock(); + + /** + * + * @return the maximum batch size used when batching writes + */ + int getMaxBatchSize(); } Modified: trunk/src/main/org/hornetq/core/protocol/core/impl/ChannelImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/protocol/core/impl/ChannelImpl.java 2010-03-22 23:43:26 UTC (rev 8946) +++ trunk/src/main/org/hornetq/core/protocol/core/impl/ChannelImpl.java 2010-03-23 15:11:11 UTC (rev 8947) @@ -29,7 +29,6 @@ import org.hornetq.core.protocol.core.Packet; import org.hornetq.core.protocol.core.impl.wireformat.HornetQExceptionMessage; import org.hornetq.core.protocol.core.impl.wireformat.PacketsConfirmedMessage; -import org.hornetq.spi.core.protocol.RemotingConnection; /** * A ChannelImpl @@ -73,9 +72,9 @@ private int receivedBytes; private CommandConfirmationHandler commandConfirmationHandler; - + private volatile boolean transferring; - + public ChannelImpl(final CoreRemotingConnection connection, final long id, final int confWindowSize) { this.connection = connection; @@ -135,30 +134,36 @@ public void sendAndFlush(final Packet packet) { - send(packet, true); + send(packet, true, false); } public void send(final Packet packet) { - send(packet, false); + send(packet, false, false); } - + + public void sendBatched(final Packet packet) + { + send(packet, false, true); + } + public void setTransferring(boolean transferring) { this.transferring = transferring; } // This must never called by more than one thread concurrently - public void send(final Packet packet, final boolean flush) + + public void send(final Packet packet, final boolean flush, final boolean batch) { synchronized (sendLock) { packet.setChannelID(id); - - final HornetQBuffer buffer = packet.encode(connection); + HornetQBuffer buffer = packet.encode(connection); + lock.lock(); - + try { while (failingOver) @@ -172,13 +177,12 @@ { } } - - //Sanity check + + // Sanity check if (transferring) { throw new IllegalStateException("Cannot send a packet while channel is doing failover"); } - if (resendCache != null && packet.isRequiresConfirmations()) { @@ -194,6 +198,10 @@ } } + public void checkFlushBatchBuffer() + { + } + public Packet sendBlocking(final Packet packet) throws HornetQException { if (closed) @@ -205,13 +213,13 @@ { throw new IllegalStateException("Cannot do a blocking call timeout on a server side connection"); } - + // Synchronized since can't be called concurrently by more than one thread and this can occur // E.g. blocking acknowledge() from inside a message handler at some time as other operation on main thread synchronized (sendBlockingLock) { packet.setChannelID(id); - + final HornetQBuffer buffer = packet.encode(connection); lock.lock(); @@ -315,7 +323,7 @@ closed = true; } - + public void transferConnection(final CoreRemotingConnection newConnection) { // Needs to synchronize on the connection to make sure no packets from @@ -373,10 +381,12 @@ { return connection; } - - //Needs to be synchronized since can be called by remoting service timer thread too for timeout flush + + // Needs to be synchronized since can be called by remoting service timer thread too for timeout flush public synchronized void flushConfirmations() { + checkFlushBatchBuffer(); + if (resendCache != null && receivedBytes != 0) { receivedBytes = 0; @@ -421,7 +431,7 @@ resendCache.clear(); } } - + public void handlePacket(final Packet packet) { if (packet.getType() == PacketImpl.PACKETS_CONFIRMED) Modified: trunk/src/main/org/hornetq/core/protocol/core/impl/CoreProtocolManager.java =================================================================== --- trunk/src/main/org/hornetq/core/protocol/core/impl/CoreProtocolManager.java 2010-03-22 23:43:26 UTC (rev 8946) +++ trunk/src/main/org/hornetq/core/protocol/core/impl/CoreProtocolManager.java 2010-03-23 15:11:11 UTC (rev 8947) @@ -43,11 +43,10 @@ public class CoreProtocolManager implements ProtocolManager { private final HornetQServer server; - + private final List<Interceptor> interceptors; - public CoreProtocolManager(final HornetQServer server, - final List<Interceptor> interceptors) + public CoreProtocolManager(final HornetQServer server, final List<Interceptor> interceptors) { this.server = server; @@ -57,11 +56,13 @@ public ConnectionEntry createConnectionEntry(final Connection connection) { final Configuration config = server.getConfiguration(); - + CoreRemotingConnection rc = new RemotingConnectionImpl(connection, interceptors, - config.isAsyncConnectionExecutionEnabled() ? server.getExecutorFactory().getExecutor() - : null); + config.isAsyncConnectionExecutionEnabled() ? server.getExecutorFactory() + .getExecutor() + : null, + connection.getBatchingBufferSize()); Channel channel1 = rc.getChannel(1, -1); @@ -102,20 +103,19 @@ return entry; } - - private Map<String, ServerSessionPacketHandler> sessionHandlers = - new ConcurrentHashMap<String, ServerSessionPacketHandler>(); - + + private Map<String, ServerSessionPacketHandler> sessionHandlers = new ConcurrentHashMap<String, ServerSessionPacketHandler>(); + public ServerSessionPacketHandler getSessionHandler(final String sessionName) { return sessionHandlers.get(sessionName); } - + public void addSessionHandler(final String name, final ServerSessionPacketHandler handler) { sessionHandlers.put(name, handler); } - + public void removeHandler(final String name) { sessionHandlers.remove(name); @@ -125,8 +125,8 @@ { } - //This is never called using the core protocol, since we override the HornetQFrameDecoder with our core - //optimised version HornetQFrameDecoder2, which nevers calls this + // This is never called using the core protocol, since we override the HornetQFrameDecoder with our core + // optimised version HornetQFrameDecoder2, which nevers calls this public int isReadyToHandle(HornetQBuffer buffer) { return -1; Modified: trunk/src/main/org/hornetq/core/protocol/core/impl/CoreSessionCallback.java =================================================================== --- trunk/src/main/org/hornetq/core/protocol/core/impl/CoreSessionCallback.java 2010-03-22 23:43:26 UTC (rev 8946) +++ trunk/src/main/org/hornetq/core/protocol/core/impl/CoreSessionCallback.java 2010-03-23 15:11:11 UTC (rev 8947) @@ -13,6 +13,7 @@ package org.hornetq.core.protocol.core.impl; import org.hornetq.api.core.SimpleString; +import org.hornetq.core.logging.Logger; import org.hornetq.core.protocol.core.Channel; import org.hornetq.core.protocol.core.Packet; import org.hornetq.core.protocol.core.impl.wireformat.SessionProducerCreditsMessage; @@ -32,12 +33,14 @@ */ public final class CoreSessionCallback implements SessionCallback { + private static final Logger log = Logger.getLogger(CoreSessionCallback.class); + private final Channel channel; private ProtocolManager protocolManager; private String name; - + public CoreSessionCallback(String name, ProtocolManager protocolManager, Channel channel) { this.name = name; @@ -51,7 +54,9 @@ channel.send(packet); - return packet.getPacketSize(); + int size = packet.getPacketSize(); + + return size; } public int sendLargeMessageContinuation(long consumerID, byte[] body, boolean continues, boolean requiresResponse) @@ -62,19 +67,21 @@ return packet.getPacketSize(); } - + public int sendMessage(ServerMessage message, long consumerID, int deliveryCount) { Packet packet = new SessionReceiveMessage(consumerID, message, deliveryCount); - channel.send(packet); + channel.sendBatched(packet); + + int size = packet.getPacketSize(); - return packet.getPacketSize(); + return size; } - public void sendProducerCreditsMessage(int credits, SimpleString address, int offset) + public void sendProducerCreditsMessage(int credits, SimpleString address) { - Packet packet = new SessionProducerCreditsMessage(credits, address, offset); + Packet packet = new SessionProducerCreditsMessage(credits, address); channel.send(packet); } Modified: trunk/src/main/org/hornetq/core/protocol/core/impl/HornetQPacketHandler.java =================================================================== --- trunk/src/main/org/hornetq/core/protocol/core/impl/HornetQPacketHandler.java 2010-03-22 23:43:26 UTC (rev 8946) +++ trunk/src/main/org/hornetq/core/protocol/core/impl/HornetQPacketHandler.java 2010-03-23 15:11:11 UTC (rev 8947) @@ -181,6 +181,7 @@ } catch (HornetQException e) { + log.error("Failed to create session ", e); response = new HornetQExceptionMessage((HornetQException)e); if (e.getCode() == HornetQException.INCOMPATIBLE_CLIENT_SERVER_VERSIONS) @@ -189,7 +190,9 @@ } } catch (Exception e) - { + { + log.error("Failed to create session ", e); + HornetQPacketHandler.log.error("Failed to create session", e); response = new HornetQExceptionMessage(new HornetQException(HornetQException.INTERNAL_ERROR)); Modified: trunk/src/main/org/hornetq/core/protocol/core/impl/RemotingConnectionImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/protocol/core/impl/RemotingConnectionImpl.java 2010-03-22 23:43:26 UTC (rev 8946) +++ trunk/src/main/org/hornetq/core/protocol/core/impl/RemotingConnectionImpl.java 2010-03-23 15:11:11 UTC (rev 8947) @@ -23,6 +23,7 @@ import java.util.concurrent.Executor; import org.hornetq.api.core.HornetQBuffer; +import org.hornetq.api.core.HornetQBuffers; import org.hornetq.api.core.HornetQException; import org.hornetq.api.core.Interceptor; import org.hornetq.core.logging.Logger; @@ -33,6 +34,7 @@ import org.hornetq.core.remoting.FailureListener; import org.hornetq.spi.core.remoting.BufferHandler; import org.hornetq.spi.core.remoting.Connection; +import org.hornetq.utils.DataConstants; import org.hornetq.utils.SimpleIDGenerator; /** @@ -87,6 +89,8 @@ private final Executor executor; + private final int maxBatchSize; + private volatile boolean executing; // Constructors @@ -99,7 +103,7 @@ final long blockingCallTimeout, final List<Interceptor> interceptors) { - this(transportConnection, blockingCallTimeout, interceptors, true, null); + this(transportConnection, blockingCallTimeout, interceptors, true, null, -1); } /* @@ -107,17 +111,19 @@ */ public RemotingConnectionImpl(final Connection transportConnection, final List<Interceptor> interceptors, - final Executor executor) + final Executor executor, + final int maxBatchSize) { - this(transportConnection, -1, interceptors, false, executor); + this(transportConnection, -1, interceptors, false, executor, maxBatchSize); } private RemotingConnectionImpl(final Connection transportConnection, final long blockingCallTimeout, final List<Interceptor> interceptors, final boolean client, - final Executor executor) + final Executor executor, + final int maxBatchSize) { this.transportConnection = transportConnection; @@ -129,6 +135,8 @@ this.client = client; this.executor = executor; + + this.maxBatchSize = maxBatchSize; } // RemotingConnection implementation @@ -363,49 +371,61 @@ } } } + + public int getMaxBatchSize() + { + return maxBatchSize; + } // Buffer Handler implementation // ---------------------------------------------------- public void bufferReceived(final Object connectionID, final HornetQBuffer buffer) { - final Packet packet = decoder.decode(buffer); - - if (packet.isAsyncExec() && executor != null) + try { - executing = true; - - executor.execute(new Runnable() + final Packet packet = decoder.decode(buffer); + + if (packet.isAsyncExec() && executor != null) { - public void run() + executing = true; + + executor.execute(new Runnable() { - try + public void run() { - doBufferReceived(packet); + try + { + doBufferReceived(packet); + } + catch (Throwable t) + { + RemotingConnectionImpl.log.error("Unexpected error", t); + } + + executing = false; } - catch (Throwable t) - { - RemotingConnectionImpl.log.error("Unexpected error", t); - } - - executing = false; + }); + } + else + { + //To prevent out of order execution if interleaving sync and async operations on same connection + while (executing) + { + Thread.yield(); } - }); + + // Pings must always be handled out of band so we can send pings back to the client quickly + // otherwise they would get in the queue with everything else which might give an intolerable delay + doBufferReceived(packet); + } + + dataReceived = true; } - else + catch (Exception e) { - //To prevent out of order execution if interleaving sync and async operations on same connection - while (executing) - { - Thread.yield(); - } - - // Pings must always be handled out of band so we can send pings back to the client quickly - // otherwise they would get in the queue with everything else which might give an intolerable delay - doBufferReceived(packet); + log.error("Failed to decode", e); } - - dataReceived = true; } private void doBufferReceived(final Packet packet) Modified: trunk/src/main/org/hornetq/core/protocol/core/impl/wireformat/SessionProducerCreditsMessage.java =================================================================== --- trunk/src/main/org/hornetq/core/protocol/core/impl/wireformat/SessionProducerCreditsMessage.java 2010-03-22 23:43:26 UTC (rev 8946) +++ trunk/src/main/org/hornetq/core/protocol/core/impl/wireformat/SessionProducerCreditsMessage.java 2010-03-23 15:11:11 UTC (rev 8947) @@ -28,24 +28,20 @@ // Attributes ---------------------------------------------------- private int credits; - + private SimpleString address; - private int offset; - // Static -------------------------------------------------------- // Constructors -------------------------------------------------- - public SessionProducerCreditsMessage(final int credits, final SimpleString address, final int offset) + public SessionProducerCreditsMessage(final int credits, final SimpleString address) { super(PacketImpl.SESS_PRODUCER_CREDITS); this.credits = credits; - + this.address = address; - - this.offset = offset; } public SessionProducerCreditsMessage() @@ -65,17 +61,11 @@ return address; } - public int getOffset() - { - return offset; - } - @Override public void encodeRest(final HornetQBuffer buffer) { buffer.writeInt(credits); buffer.writeSimpleString(address); - buffer.writeInt(offset); } @Override @@ -83,7 +73,6 @@ { credits = buffer.readInt(); address = buffer.readSimpleString(); - offset = buffer.readInt(); } // Package protected --------------------------------------------- Modified: trunk/src/main/org/hornetq/core/protocol/core/impl/wireformat/SessionReceiveLargeMessage.java =================================================================== --- trunk/src/main/org/hornetq/core/protocol/core/impl/wireformat/SessionReceiveLargeMessage.java 2010-03-22 23:43:26 UTC (rev 8946) +++ trunk/src/main/org/hornetq/core/protocol/core/impl/wireformat/SessionReceiveLargeMessage.java 2010-03-23 15:11:11 UTC (rev 8947) @@ -14,6 +14,7 @@ package org.hornetq.core.protocol.core.impl.wireformat; import org.hornetq.api.core.HornetQBuffer; +import org.hornetq.core.logging.Logger; import org.hornetq.core.protocol.core.impl.PacketImpl; /** @@ -25,6 +26,8 @@ */ public class SessionReceiveLargeMessage extends PacketImpl { + private static final Logger log = Logger.getLogger(SessionReceiveLargeMessage.class); + private byte[] largeMessageHeader; /** Since we receive the message before the entire message was received, */ @@ -45,7 +48,7 @@ final int deliveryCount) { super(PacketImpl.SESS_RECEIVE_LARGE_MSG); - + this.consumerID = consumerID; this.largeMessageHeader = largeMessageHeader; Modified: trunk/src/main/org/hornetq/core/protocol/stomp/StompSession.java =================================================================== --- trunk/src/main/org/hornetq/core/protocol/stomp/StompSession.java 2010-03-22 23:43:26 UTC (rev 8946) +++ trunk/src/main/org/hornetq/core/protocol/stomp/StompSession.java 2010-03-23 15:11:11 UTC (rev 8947) @@ -72,7 +72,7 @@ return session; } - public void sendProducerCreditsMessage(int credits, SimpleString address, int offset) + public void sendProducerCreditsMessage(int credits, SimpleString address) { } Modified: trunk/src/main/org/hornetq/core/protocol/stomp/StompUtils.java =================================================================== --- trunk/src/main/org/hornetq/core/protocol/stomp/StompUtils.java 2010-03-22 23:43:26 UTC (rev 8946) +++ trunk/src/main/org/hornetq/core/protocol/stomp/StompUtils.java 2010-03-23 15:11:11 UTC (rev 8947) @@ -21,6 +21,7 @@ import org.hornetq.api.core.Message; import org.hornetq.api.core.SimpleString; import org.hornetq.core.client.impl.ClientMessageImpl; +import org.hornetq.core.message.impl.MessageInternal; import org.hornetq.core.server.impl.ServerMessageImpl; /** @@ -82,7 +83,7 @@ } } - public static void copyStandardHeadersFromMessageToFrame(Message message, StompFrame command, int deliveryCount) throws Exception + public static void copyStandardHeadersFromMessageToFrame(MessageInternal message, StompFrame command, int deliveryCount) throws Exception { final Map<String, Object> headers = command.getHeaders(); headers.put(Stomp.Headers.Message.DESTINATION, message.getAddress().toString()); Modified: trunk/src/main/org/hornetq/core/remoting/impl/invm/InVMConnection.java =================================================================== --- trunk/src/main/org/hornetq/core/remoting/impl/invm/InVMConnection.java 2010-03-22 23:43:26 UTC (rev 8946) +++ trunk/src/main/org/hornetq/core/remoting/impl/invm/InVMConnection.java 2010-03-23 15:11:11 UTC (rev 8947) @@ -150,5 +150,10 @@ { return "invm:" + serverID; } + + public int getBatchingBufferSize() + { + return -1; + } } Modified: trunk/src/main/org/hornetq/core/remoting/server/impl/RemotingServiceImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/remoting/server/impl/RemotingServiceImpl.java 2010-03-22 23:43:26 UTC (rev 8946) +++ trunk/src/main/org/hornetq/core/remoting/server/impl/RemotingServiceImpl.java 2010-03-23 15:11:11 UTC (rev 8947) @@ -19,8 +19,11 @@ import java.util.Set; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.CopyOnWriteArrayList; -import java.util.concurrent.Executor; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ThreadFactory; +import java.util.concurrent.TimeUnit; import org.hornetq.api.core.HornetQBuffer; import org.hornetq.api.core.HornetQException; @@ -45,6 +48,7 @@ import org.hornetq.spi.core.remoting.Connection; import org.hornetq.spi.core.remoting.ConnectionLifeCycleListener; import org.hornetq.utils.ConfigurationHelper; +import org.hornetq.utils.HornetQThreadFactory; /** * @author <a href="mailto:jmesnil@redhat.com">Jeff Mesnil</a> @@ -80,7 +84,7 @@ private volatile RemotingConnection serverSideReplicatingConnection; - private final Executor threadPool; + private ExecutorService threadPool; private final ScheduledExecutorService scheduledThreadPool; @@ -95,8 +99,7 @@ public RemotingServiceImpl(final Configuration config, final HornetQServer server, - final ManagementService managementService, - final Executor threadPool, + final ManagementService managementService, final ScheduledExecutorService scheduledThreadPool) { transportConfigs = config.getAcceptorConfigurations(); @@ -119,7 +122,7 @@ this.config = config; this.managementService = managementService; - this.threadPool = threadPool; + this.scheduledThreadPool = scheduledThreadPool; this.protocolMap.put(ProtocolType.CORE, new CoreProtocolManagerFactory().createProtocolManager(server, interceptors)); @@ -135,7 +138,17 @@ { return; } + + //The remoting service maintains it's own thread pool for handling remoting traffic + //If OIO each connection will have it's own thread + //If NIO these are capped at nio-remoting-threads which defaults to num cores * 3 + //This needs to be a different thread pool to the main thread pool especially for OIO where we may need + //to support many hundreds of connections, but the main thread pool must be kept small for better performance + + ThreadFactory tFactory = new HornetQThreadFactory("HornetQ-remoting-threads" + System.identityHashCode(this), false); + threadPool = Executors.newCachedThreadPool(tFactory); + ClassLoader loader = Thread.currentThread().getContextClassLoader(); for (TransportConfiguration info : transportConfigs) @@ -262,6 +275,15 @@ { managementService.unregisterAcceptors(); } + + threadPool.shutdown(); + + boolean ok = threadPool.awaitTermination(10000, TimeUnit.MILLISECONDS); + + if (!ok) + { + log.warn("Timed out waiting for remoting thread pool to terminate"); + } started = false; Modified: trunk/src/main/org/hornetq/core/replication/impl/ReplicationEndpointImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/replication/impl/ReplicationEndpointImpl.java 2010-03-22 23:43:26 UTC (rev 8946) +++ trunk/src/main/org/hornetq/core/replication/impl/ReplicationEndpointImpl.java 2010-03-23 15:11:11 UTC (rev 8947) @@ -389,6 +389,7 @@ private void handleLargeMessageBegin(final ReplicationLargeMessageBeingMessage packet) { LargeServerMessage largeMessage = storage.createLargeMessage(); + largeMessage.setDurable(true); largeMessage.setMessageID(packet.getMessageId()); ReplicationEndpointImpl.trace("Receiving Large Message " + largeMessage.getMessageID() + " on backup"); largeMessages.put(largeMessage.getMessageID(), largeMessage); Modified: trunk/src/main/org/hornetq/core/server/Queue.java =================================================================== --- trunk/src/main/org/hornetq/core/server/Queue.java 2010-03-22 23:43:26 UTC (rev 8946) +++ trunk/src/main/org/hornetq/core/server/Queue.java 2010-03-23 15:11:11 UTC (rev 8947) @@ -45,7 +45,7 @@ void addConsumer(Consumer consumer) throws Exception; - boolean removeConsumer(Consumer consumer) throws Exception; + void removeConsumer(Consumer consumer) throws Exception; int getConsumerCount(); @@ -75,12 +75,8 @@ List<MessageReference> getScheduledMessages(); - // Distributor getDistributionPolicy(); - // - // void setDistributionPolicy(Distributor policy); + long getMessagesAdded(); - int getMessagesAdded(); - MessageReference removeReferenceWithID(long id) throws Exception; MessageReference removeFirstReference(long id) throws Exception; Modified: trunk/src/main/org/hornetq/core/server/ServerMessage.java =================================================================== --- trunk/src/main/org/hornetq/core/server/ServerMessage.java 2010-03-22 23:43:26 UTC (rev 8946) +++ trunk/src/main/org/hornetq/core/server/ServerMessage.java 2010-03-23 15:11:11 UTC (rev 8947) @@ -31,9 +31,9 @@ MessageReference createReference(Queue queue); - int incrementRefCount(MessageReference reference) throws Exception; + int incrementRefCount() throws Exception; - int decrementRefCount(MessageReference reference) throws Exception; + int decrementRefCount() throws Exception; int incrementDurableRefCount(); Modified: trunk/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java 2010-03-22 23:43:26 UTC (rev 8946) +++ trunk/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java 2010-03-23 15:11:11 UTC (rev 8947) @@ -40,8 +40,8 @@ import org.hornetq.core.client.impl.FailoverManager; import org.hornetq.core.client.impl.FailoverManagerImpl; import org.hornetq.core.config.Configuration; -import org.hornetq.core.config.DivertConfiguration; import org.hornetq.core.config.CoreQueueConfiguration; +import org.hornetq.core.config.DivertConfiguration; import org.hornetq.core.config.impl.ConfigurationImpl; import org.hornetq.core.deployers.Deployer; import org.hornetq.core.deployers.DeploymentManager; @@ -554,6 +554,7 @@ final boolean xa, final SessionCallback callback) throws Exception { + if (securityStore != null) { securityStore.authenticate(username, password); @@ -790,12 +791,6 @@ addressSettingsRepository); } - /** for use on sub-classes */ - protected ExecutorService getExecutor() - { - return threadPool; - } - /** * This method is protected as it may be used as a hook for creating a custom storage manager (on tests for instance) */ @@ -902,7 +897,7 @@ managementService = new ManagementServiceImpl(mbeanServer, configuration); - remotingService = new RemotingServiceImpl(configuration, this, managementService, threadPool, scheduledPool); + remotingService = new RemotingServiceImpl(configuration, this, managementService, scheduledPool); if (configuration.getMemoryMeasureInterval() != -1) { Modified: trunk/src/main/org/hornetq/core/server/impl/QueueImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/server/impl/QueueImpl.java 2010-03-22 23:43:26 UTC (rev 8946) +++ trunk/src/main/org/hornetq/core/server/impl/QueueImpl.java 2010-03-23 15:11:11 UTC (rev 8947) @@ -21,20 +21,16 @@ import java.util.List; import java.util.Map; import java.util.Set; -import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.ConcurrentMap; import java.util.concurrent.Executor; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.ScheduledFuture; import java.util.concurrent.TimeUnit; -import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicLong; import org.hornetq.api.core.Message; import org.hornetq.api.core.SimpleString; import org.hornetq.core.filter.Filter; -import org.hornetq.core.list.PriorityLinkedList; -import org.hornetq.core.list.impl.PriorityLinkedListImpl; import org.hornetq.core.logging.Logger; import org.hornetq.core.persistence.StorageManager; import org.hornetq.core.postoffice.Bindings; @@ -54,6 +50,9 @@ import org.hornetq.core.transaction.TransactionPropertyIndexes; import org.hornetq.core.transaction.impl.TransactionImpl; import org.hornetq.utils.ConcurrentHashSet; +import org.hornetq.utils.PriorityLinkedList; +import org.hornetq.utils.PriorityLinkedListImpl; +import org.hornetq.utils.concurrent.HQIterator; /** * Implementation of a Queue @@ -83,22 +82,16 @@ private final PostOffice postOffice; - private final PriorityLinkedList<MessageReference> messageReferences = new PriorityLinkedListImpl<MessageReference>(QueueImpl.NUM_PRIORITIES); + private final PriorityLinkedList<MessageReference> messageReferences = new PriorityLinkedListImpl<MessageReference>(true, QueueImpl.NUM_PRIORITIES); - private final List<MessageHandler> handlers = new ArrayList<MessageHandler>(); + private final List<ConsumerHolder> consumerList = new ArrayList<ConsumerHolder>(); private final ScheduledDeliveryHandler scheduledDeliveryHandler; - private boolean direct; + private final AtomicLong messagesAdded = new AtomicLong(0); - private boolean promptDelivery; - - private final AtomicInteger messagesAdded = new AtomicInteger(0); - protected final AtomicInteger deliveringCount = new AtomicInteger(0); - private final AtomicBoolean waitingToDeliver = new AtomicBoolean(false); - private boolean paused; private final Runnable deliverRunner = new DeliverRunner(); @@ -121,7 +114,7 @@ private final Set<Consumer> consumerSet = new HashSet<Consumer>(); - private final ConcurrentMap<SimpleString, Consumer> groups = new ConcurrentHashMap<SimpleString, Consumer>(); + private final Map<SimpleString, Consumer> groups = new HashMap<SimpleString, Consumer>(); private volatile SimpleString expiryAddress; @@ -129,6 +122,18 @@ private final Executor executor; + private static class ConsumerHolder + { + ConsumerHolder(final Consumer consumer) + { + this.consumer = consumer; + } + + final Consumer consumer; + + volatile HQIterator<MessageReference> iter; + } + public QueueImpl(final long id, final SimpleString address, final SimpleString name, @@ -161,8 +166,6 @@ this.scheduledExecutor = scheduledExecutor; - direct = true; - scheduledDeliveryHandler = new ScheduledDeliveryHandlerImpl(scheduledExecutor); if (addressSettingsRepository != null) @@ -228,6 +231,8 @@ public void addLast(final MessageReference ref) { + messagesAdded.incrementAndGet(); + add(ref, false); } @@ -238,12 +243,7 @@ public void deliverAsync() { - // Prevent too many executors running at once - - if (waitingToDeliver.compareAndSet(false, true)) - { - executor.execute(deliverRunner); - } + executor.execute(deliverRunner); } public Executor getExecutor() @@ -253,52 +253,55 @@ public synchronized void deliverNow() { - deliverRunner.run(); + deliver(); } public synchronized void addConsumer(final Consumer consumer) throws Exception { cancelRedistributor(); - MessageHandler handler; + consumerList.add(new ConsumerHolder(consumer)); - if (consumer.getFilter() != null) - { - handler = new FilterMessageHandler(consumer, messageReferences.iterator()); - } - else - { - handler = new NullFilterMessageHandler(consumer); - } - - handlers.add(handler); - consumerSet.add(consumer); } - public synchronized boolean removeConsumer(final Consumer consumer) throws Exception + public synchronized void removeConsumer(final Consumer consumer) throws Exception { - boolean removed = removeHandlerGivenConsumer(consumer); - - if (handlers.isEmpty()) + Iterator<ConsumerHolder> iter = consumerList.iterator(); + + while (iter.hasNext()) { - promptDelivery = false; + ConsumerHolder holder = iter.next(); + + if (holder.consumer == consumer) + { + iter.remove(); + + break; + } } + + if (pos > 0 && pos >= consumerList.size()) + { + pos = consumerList.size() - 1; + } consumerSet.remove(consumer); - if (removed) + List<SimpleString> gids = new ArrayList<SimpleString>(); + + for (SimpleString groupID : groups.keySet()) { - for (SimpleString groupID : groups.keySet()) + if (consumer == groups.get(groupID)) { - if (consumer == groups.get(groupID)) - { - groups.remove(groupID); - } + gids.add(groupID); } } - - return removed; + + for (SimpleString gid : gids) + { + groups.remove(gid); + } } public synchronized void addRedistributor(final long delay) @@ -341,7 +344,24 @@ redistributor = null; - removeHandlerGivenConsumer(redistributor); + Iterator<ConsumerHolder> iter = consumerList.iterator(); + + while (iter.hasNext()) + { + ConsumerHolder holder = iter.next(); + + if (holder.consumer == redistributor) + { + iter.remove(); + + break; + } + } + + if (pos > 0 && pos >= consumerList.size()) + { + pos = consumerList.size() - 1; + } } if (future != null) @@ -364,9 +384,9 @@ public synchronized boolean hasMatchingConsumer(final ServerMessage message) { - for (MessageHandler handler : handlers) + for (ConsumerHolder holder : consumerList) { - Consumer consumer = handler.getConsumer(); + Consumer consumer = holder.consumer; if (consumer instanceof Redistributor) { @@ -394,28 +414,39 @@ { return new Iterator<MessageReference>() { - private final Iterator<MessageReference> iterator = messageReferences.iterator(); + private final HQIterator<MessageReference> iterator = messageReferences.iterator(); + private MessageReference next; + public boolean hasNext() { - return iterator.hasNext(); + if (next == null) + { + next = iterator.next(); + } + + return next != null; } public MessageReference next() { - return iterator.next(); + MessageReference n = next; + + next = null; + + return n; } public void remove() { - throw new UnsupportedOperationException("iterator is immutable"); + iterator.remove(); } }; } public MessageReference removeReferenceWithID(final long id) throws Exception { - Iterator<MessageReference> iterator = messageReferences.iterator(); + Iterator<MessageReference> iterator = iterator(); MessageReference removed = null; @@ -462,7 +493,7 @@ public MessageReference getReference(final long id) { - Iterator<MessageReference> iterator = messageReferences.iterator(); + Iterator<MessageReference> iterator = iterator(); while (iterator.hasNext()) { @@ -596,7 +627,7 @@ deliveringCount.incrementAndGet(); } - public int getMessagesAdded() + public long getMessagesAdded() { return messagesAdded.get(); } @@ -612,7 +643,7 @@ Transaction tx = new TransactionImpl(storageManager); - Iterator<MessageReference> iter = messageReferences.iterator(); + Iterator<MessageReference> iter = iterator(); while (iter.hasNext()) { @@ -646,7 +677,7 @@ Transaction tx = new TransactionImpl(storageManager); - Iterator<MessageReference> iter = messageReferences.iterator(); + Iterator<MessageReference> iter = iterator(); while (iter.hasNext()) { @@ -668,7 +699,7 @@ public boolean expireReference(final long messageID) throws Exception { - Iterator<MessageReference> iter = messageReferences.iterator(); + Iterator<MessageReference> iter = iterator(); while (iter.hasNext()) { @@ -689,7 +720,7 @@ Transaction tx = new TransactionImpl(storageManager); int count = 0; - Iterator<MessageReference> iter = messageReferences.iterator(); + Iterator<MessageReference> iter = iterator(); while (iter.hasNext()) { @@ -710,7 +741,7 @@ public void expireReferences() throws Exception { - Iterator<MessageReference> iter = messageReferences.iterator(); + Iterator<MessageReference> iter = iterator(); while (iter.hasNext()) { @@ -726,7 +757,7 @@ public boolean sendMessageToDeadLetterAddress(final long messageID) throws Exception { - Iterator<MessageReference> iter = messageReferences.iterator(); + Iterator<MessageReference> iter = iterator(); while (iter.hasNext()) { @@ -745,7 +776,7 @@ public int sendMessagesToDeadLetterAddress(Filter filter) throws Exception { int count = 0; - Iterator<MessageReference> iter = messageReferences.iterator(); + Iterator<MessageReference> iter = iterator(); while (iter.hasNext()) { @@ -763,7 +794,7 @@ public boolean moveReference(final long messageID, final SimpleString toAddress) throws Exception { - Iterator<MessageReference> iter = messageReferences.iterator(); + Iterator<MessageReference> iter = iterator(); while (iter.hasNext()) { @@ -784,7 +815,7 @@ Transaction tx = new TransactionImpl(storageManager); int count = 0; - Iterator<MessageReference> iter = messageReferences.iterator(); + Iterator<MessageReference> iter = iterator(); while (iter.hasNext()) { @@ -814,7 +845,7 @@ public boolean changeReferencePriority(final long messageID, final byte newPriority) throws Exception { - Iterator<MessageReference> iter = messageReferences.iterator(); + Iterator<MessageReference> iter = iterator(); while (iter.hasNext()) { @@ -833,7 +864,7 @@ public int changeReferencesPriority(final Filter filter, final byte newPriority) throws Exception { - Iterator<MessageReference> iter = messageReferences.iterator(); + Iterator<MessageReference> iter = iterator(); int count = 0; while (iter.hasNext()) @@ -881,33 +912,6 @@ // Private // ------------------------------------------------------------------------------ - private boolean removeHandlerGivenConsumer(final Consumer consumer) - { - Iterator<MessageHandler> iter = handlers.iterator(); - - boolean removed = false; - - while (iter.hasNext()) - { - MessageHandler handler = iter.next(); - - if (handler.getConsumer() == consumer) - { - iter.remove(); - - if (pos >= handlers.size()) - { - pos = 0; - } - removed = true; - - break; - } - } - - return removed; - } - private void internalAddRedistributor(final Executor executor) { // create the redistributor only once if there are no local consumers @@ -919,7 +923,7 @@ executor, QueueImpl.REDISTRIBUTOR_BATCH_SIZE); - handlers.add(new NullFilterMessageHandler(redistributor)); + consumerList.add(new ConsumerHolder(redistributor)); redistributor.start(); @@ -1073,269 +1077,204 @@ tx.commit(); } - private MessageHandler getHandlerRoundRobin() + private synchronized void deliver() { - MessageHandler handler = handlers.get(pos); - - pos++; - - if (pos == handlers.size()) + if (paused || consumerList.isEmpty()) { - pos = 0; + return; } + + // Disadvantage of this algorithm is that if there are many consumers which are busy a lot of the + // time, then they get tried with a message each time, and the message put back on the queue, which + // is inefficient - return handler; - } + // This represents the number of consumers that are unavailable to take a message due either to + // there not being any messages available for its iterator/in queue or it's busy + // int unavailableCount = 0; - private boolean checkExpired(final MessageReference reference) - { - if (reference.getMessage().isExpired()) - { - reference.handled(); + int busyCount = 0; - try - { - expire(reference); - } - catch (Exception e) - { - QueueImpl.log.error("Failed to expire ref", e); - } + int nullRefCount = 0; - return true; - } - else - { - return false; - } - } + int size = consumerList.size(); - /* - * Attempt to deliver all the messages in the queue - */ - private synchronized void deliver() - { - if (paused || handlers.isEmpty()) - { - return; - } - - direct = false; - int startPos = pos; - int totalCount = handlers.size(); - int nullCount = 0; - int busyCount = 0; - while (true) + + // Deliver at most 1000 messages in one go, to prevent tying this thread up for too long + int loop = Math.min(messageReferences.size(), 1000); + + for (int i = 0; i < loop; i++) { - MessageHandler handler = getHandlerRoundRobin(); + ConsumerHolder holder = consumerList.get(pos); - Consumer consumer = handler.getConsumer(); + Consumer consumer = holder.consumer; - MessageReference reference = handler.peek(consumer); + MessageReference ref; - if (reference == null) + if (holder.iter == null) { - nullCount++; + ref = messageReferences.removeFirst(); } else { - if (checkExpired(reference)) - { - handler.remove(); - } - else - { - final SimpleString groupID = reference.getMessage().getSimpleStringProperty(Message.HDR_GROUP_ID); - - boolean tryHandle = true; - - if (groupID != null) - { - Consumer groupConsumer = groups.putIfAbsent(groupID, consumer); - - if (groupConsumer != null && groupConsumer != consumer) - { - tryHandle = false; - - busyCount++; - } - } - - if (tryHandle) - { - HandleStatus status = handle(reference, consumer); - - if (status == HandleStatus.HANDLED) - { - handler.remove(); - } - else if (status == HandleStatus.BUSY) - { - busyCount++; - - handler.reset(); - } - else if (status == HandleStatus.NO_MATCH) - { - // if consumer filter reject the message make sure it won't be assigned the message group - if (groupID != null) - { - groups.remove(groupID); - } - } - } - } + ref = holder.iter.next(); } - if (pos == startPos) + if (ref == null) { - // We've done all the consumers - - if (nullCount + busyCount == totalCount) + nullRefCount++; + } + else + { + if (checkExpired(ref)) { - if (nullCount == totalCount) + if (holder.iter != null) { - // We delivered all the messages - go into direct delivery - direct = true; - - promptDelivery = false; + holder.iter.remove(); } - break; + continue; } - nullCount = busyCount = 0; - } - } - } + Consumer groupConsumer = null; + + //If a group id is set, then this overrides the consumer chosen round-robin + + SimpleString groupID = ref.getMessage().getSimpleStringProperty(Message.HDR_GROUP_ID); - private synchronized boolean directDeliver(final MessageReference reference) - { - if (paused || handlers.isEmpty()) - { - return false; - } - - if (checkExpired(reference)) - { - return true; - } - - int startPos = pos; - int busyCount = 0; - boolean setPromptDelivery = false; - while (true) - { - MessageHandler handler = getHandlerRoundRobin(); - - Consumer consumer = handler.getConsumer(); - - SimpleString groupID = reference.getMessage().getSimpleStringProperty(Message.HDR_GROUP_ID); - - boolean tryHandle = true; - - if (groupID != null) - { - Consumer groupConsumer = groups.putIfAbsent(groupID, consumer); - - if (groupConsumer != null && groupConsumer != consumer) + if (groupID != null) { - tryHandle = false; + groupConsumer = groups.get(groupID); + + if (groupConsumer != null) + { + consumer = groupConsumer; + } } - } + + HandleStatus status = handle(ref, consumer); - if (tryHandle) - { - HandleStatus status = handle(reference, consumer); - if (status == HandleStatus.HANDLED) { - return true; + if (holder.iter != null) + { + holder.iter.remove(); + } + + if (groupID != null && groupConsumer == null) + { + groups.put(groupID, consumer); + } } else if (status == HandleStatus.BUSY) { - busyCount++; - - if (groupID != null) + if (holder.iter == null) { - return false; + // Put the ref back + + messageReferences.addFirst(ref, ref.getMessage().getPriority()); } + + busyCount++; } else if (status == HandleStatus.NO_MATCH) { - // if consumer filter reject the message make sure it won't be assigned the message group - if (groupID != null) + if (holder.iter == null) { - groups.remove(groupID); - } + // Put the ref back - setPromptDelivery = true; + messageReferences.addFirst(ref, ref.getMessage().getPriority()); + + holder.iter = messageReferences.iterator(); + + //Skip past the one we just put back + + holder.iter.next(); + } } } + + pos++; + if (pos == size) + { + pos = 0; + } + if (pos == startPos) { - if (setPromptDelivery) + // Round robin'd all + + if (nullRefCount + busyCount == size) { - promptDelivery = true; + break; } - return false; + nullRefCount = busyCount = 0; } } + + if (messageReferences.size() > 0 && busyCount != size) + { + // More messages to deliver so need to prompt another runner - note we don't + // prompt another one if all consumers are busy + + executor.execute(deliverRunner); + } + } - protected synchronized void add(final MessageReference ref, final boolean first) + private boolean checkExpired(final MessageReference reference) { - if (!first) + if (reference.getMessage().isExpired()) { - messagesAdded.incrementAndGet(); + reference.handled(); + + try + { + expire(reference); + } + catch (Exception e) + { + QueueImpl.log.error("Failed to expire ref", e); + } + + return true; } + else + { + return false; + } + } + protected void add(final MessageReference ref, final boolean first) + { if (scheduledDeliveryHandler.checkAndSchedule(ref)) { return; } - boolean add = false; - - if (direct && !paused) + int refs; + + if (first) { - // Deliver directly - - boolean delivered = directDeliver(ref); - - if (!delivered) - { - add = true; - - direct = false; - } + refs = messageReferences.addFirst(ref, ref.getMessage().getPriority()); } else { - add = true; + refs = messageReferences.addLast(ref, ref.getMessage().getPriority()); } - if (add) + /* + * We only prompt delivery if there are no messages waiting for delivery - this prevents many executors being + * unnecessarily queued up + * During delivery toDeliver is decremented before the message is delivered, therefore if it's delivering the last + * message, then we cannot have a situation where this delivery is not prompted and message remains stranded in the + * queue + */ + if (refs == 1) { - if (first) - { - messageReferences.addFirst(ref, ref.getMessage().getPriority()); - } - else - { - messageReferences.addLast(ref, ref.getMessage().getPriority()); - } - - if (!direct && promptDelivery) - { - // We have consumers with filters which don't match, so we need - // to prompt delivery every time - // a new message arrives - deliverAsync(); - } + deliverAsync(); } } @@ -1410,15 +1349,13 @@ queue.deliveringCount.decrementAndGet(); - message.decrementRefCount(ref); + message.decrementRefCount(); } void postRollback(final LinkedList<MessageReference> refs) { synchronized (this) { - direct = false; - for (MessageReference ref : refs) { add(ref, true); @@ -1431,17 +1368,6 @@ // Inner classes // -------------------------------------------------------------------------- - private class DeliverRunner implements Runnable - { - public void run() - { - // Must be set to false *before* executing to avoid race - waitingToDeliver.set(false); - - deliver(); - } - } - private final class RefsOperation implements TransactionOperation { List<MessageReference> refsToAck = new ArrayList<MessageReference>(); @@ -1557,109 +1483,18 @@ return paused; } - private static interface MessageHandler + private class DeliverRunner implements Runnable { - MessageReference peek(Consumer consumer); - - void remove(); - - void reset(); - - Consumer getConsumer(); - } - - private class FilterMessageHandler implements MessageHandler - { - private final Consumer consumer; - - private Iterator<MessageReference> iterator; - - private MessageReference lastReference; - - private boolean resetting; - - public FilterMessageHandler(final Consumer consumer, final Iterator<MessageReference> iterator) + public void run() { - this.consumer = consumer; - - this.iterator = iterator; - } - - public MessageReference peek(final Consumer consumer) - { - if (resetting) + try { - resetting = false; - - return lastReference; + deliver(); } - - MessageReference reference; - - if (iterator.hasNext()) + catch (Exception e) { - reference = iterator.next(); + log.error("Failed to deliver", e); } - else - { - reference = null; - - if (consumer.getFilter() != null) - { - // we have iterated on the whole queue for - // messages which matches the consumer filter. - // we reset its iterator in case new messages are added to the queue - iterator = messageReferences.iterator(); - } - } - lastReference = reference; - - return reference; } - - public void remove() - { - iterator.remove(); - } - - public void reset() - { - resetting = true; - } - - public Consumer getConsumer() - { - return consumer; - } } - - private class NullFilterMessageHandler implements MessageHandler - { - private final Consumer consumer; - - NullFilterMessageHandler(final Consumer consumer) - { - this.consumer = consumer; - } - - public MessageReference peek(final Consumer consumer) - { - return messageReferences.peekFirst(); - } - - public void remove() - { - messageReferences.removeFirst(); - } - - public void reset() - { - // no-op - } - - public Consumer getConsumer() - { - return consumer; - } - } } Modified: trunk/src/main/org/hornetq/core/server/impl/ServerConsumerImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/server/impl/ServerConsumerImpl.java 2010-03-22 23:43:26 UTC (rev 8946) +++ trunk/src/main/org/hornetq/core/server/impl/ServerConsumerImpl.java 2010-03-23 15:11:11 UTC (rev 8947) @@ -18,8 +18,6 @@ import java.util.concurrent.ConcurrentLinkedQueue; import java.util.concurrent.Executor; import java.util.concurrent.atomic.AtomicInteger; -import java.util.concurrent.locks.Lock; -import java.util.concurrent.locks.ReentrantLock; import org.hornetq.api.core.HornetQBuffer; import org.hornetq.api.core.HornetQBuffers; @@ -85,7 +83,7 @@ private final Executor executor; - private final Lock lock = new ReentrantLock(); + private final Object lock = new Object(); private AtomicInteger availableCredits = new AtomicInteger(0); @@ -186,10 +184,8 @@ { return HandleStatus.BUSY; } - - lock.lock(); - - try + + synchronized (lock) { // If the consumer is stopped then we don't accept the message, it // should go back into the @@ -225,7 +221,7 @@ ref.incrementDeliveryCount(); // If updateDeliveries = false (set by strict-update), - // the updateDeliveryCount would still be updated after cancel + // the updateDeliveryCount would still be updated after c if (strictUpdateDeliveryCount) { if (ref.getMessage().isDurable() && ref.getQueue().isDurable()) @@ -259,10 +255,6 @@ return HandleStatus.HANDLED; } - finally - { - lock.unlock(); - } } public Filter getFilter() @@ -396,15 +388,10 @@ public void setStarted(final boolean started) { - lock.lock(); - try + synchronized (lock) { this.started = browseOnly || started; } - finally - { - lock.unlock(); - } // Outside the lock if (started) @@ -415,8 +402,7 @@ public void setTransferring(final boolean transferring) { - lock.lock(); - try + synchronized (lock) { this.transferring = transferring; @@ -435,10 +421,6 @@ } } } - finally - { - lock.unlock(); - } // Outside the lock if (transferring) @@ -503,7 +485,7 @@ { return; } - + // Acknowledge acknowledges all refs delivered by the consumer up to and including the one explicitly // acknowledged @@ -578,8 +560,7 @@ private void promptDelivery() { - lock.lock(); - try + synchronized (lock) { // largeMessageDeliverer is aways set inside a lock // if we don't acquire a lock, we will have NPE eventually @@ -599,10 +580,6 @@ } } } - finally - { - lock.unlock(); - } } private void resumeLargeMessage() @@ -642,31 +619,29 @@ { public void run() { - lock.lock(); - try + synchronized (lock) { - if (largeMessageDeliverer == null || largeMessageDeliverer.deliver()) + try { - if (browseOnly) + if (largeMessageDeliverer == null || largeMessageDeliverer.deliver()) { - executor.execute(browserDeliverer); - } - else - { - // prompt Delivery only if chunk was finished + if (browseOnly) + { + executor.execute(browserDeliverer); + } + else + { + // prompt Delivery only if chunk was finished - messageQueue.deliverAsync(); + messageQueue.deliverAsync(); + } } } + catch (Exception e) + { + ServerConsumerImpl.log.error("Failed to run large message deliverer", e); + } } - catch (Exception e) - { - ServerConsumerImpl.log.error("Failed to run large message deliverer", e); - } - finally - { - lock.unlock(); - } } }; @@ -698,9 +673,7 @@ public boolean deliver() throws Exception { - lock.lock(); - - try + synchronized (lock) { if (largeMessage == null) { @@ -721,7 +694,7 @@ context = largeMessage.getBodyEncoder(); sizePendingLargeMessage = context.getLargeBodySize(); - + context.open(); sentInitialPacket = true; @@ -803,16 +776,11 @@ return true; } - finally - { - lock.unlock(); - } } public void finish() throws Exception { - lock.lock(); - try + synchronized (lock) { if (largeMessage == null) { @@ -839,10 +807,6 @@ largeMessage = null; } - finally - { - lock.unlock(); - } } } Modified: trunk/src/main/org/hornetq/core/server/impl/ServerInfo.java =================================================================== --- trunk/src/main/org/hornetq/core/server/impl/ServerInfo.java 2010-03-22 23:43:26 UTC (rev 8946) +++ trunk/src/main/org/hornetq/core/server/impl/ServerInfo.java 2010-03-23 15:11:11 UTC (rev 8947) @@ -83,7 +83,7 @@ private String appendPagingInfos() { String info = ""; - info += String.format("total paging memory: %s\n", SizeFormatterUtil.sizeof(pagingManager.getTotalMemory())); + for (SimpleString storeName : pagingManager.getStoreNames()) { PagingStore pageStore; Modified: trunk/src/main/org/hornetq/core/server/impl/ServerMessageImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/server/impl/ServerMessageImpl.java 2010-03-22 23:43:26 UTC (rev 8946) +++ trunk/src/main/org/hornetq/core/server/impl/ServerMessageImpl.java 2010-03-23 15:11:11 UTC (rev 8947) @@ -14,7 +14,6 @@ package org.hornetq.core.server.impl; import java.io.InputStream; -import java.util.concurrent.atomic.AtomicInteger; import org.hornetq.api.core.Message; import org.hornetq.api.core.SimpleString; @@ -40,10 +39,9 @@ { private static final Logger log = Logger.getLogger(ServerMessageImpl.class); - private final AtomicInteger durableRefCount = new AtomicInteger(0); + private int durableRefCount; - /** Global reference counts for paging control */ - private final AtomicInteger refCount = new AtomicInteger(0); + private int refCount; private PagingStore pagingStore; @@ -108,53 +106,57 @@ return ref; } - public int incrementRefCount(final MessageReference reference) throws Exception + public synchronized int incrementRefCount() throws Exception { - int count = refCount.incrementAndGet(); - + refCount ++; + if (pagingStore != null) - { - if (count == 1) + { + if (refCount == 1) { - pagingStore.addSize(this, true); + pagingStore.addSize(getMemoryEstimate() + MessageReferenceImpl.getMemoryEstimate()); } - - pagingStore.addSize(reference, true); + else + { + pagingStore.addSize(MessageReferenceImpl.getMemoryEstimate()); + } } - return count; + return refCount; } - public int decrementRefCount(final MessageReference reference) throws Exception + public synchronized int decrementRefCount() throws Exception { - int count = refCount.decrementAndGet(); + int count = --refCount; if (pagingStore != null) { if (count == 0) { - pagingStore.addSize(this, false); + pagingStore.addSize(-getMemoryEstimate() - MessageReferenceImpl.getMemoryEstimate()); } - - pagingStore.addSize(reference, false); + else + { + pagingStore.addSize(-MessageReferenceImpl.getMemoryEstimate()); + } } return count; } - public int incrementDurableRefCount() + public synchronized int incrementDurableRefCount() { - return durableRefCount.incrementAndGet(); + return ++durableRefCount; } - public int decrementDurableRefCount() + public synchronized int decrementDurableRefCount() { - return durableRefCount.decrementAndGet(); + return --durableRefCount; } - public int getRefCount() + public synchronized int getRefCount() { - return refCount.get(); + return refCount; } public boolean isLargeMessage() Deleted: trunk/src/main/org/hornetq/core/server/impl/ServerProducerCreditManager.java =================================================================== --- trunk/src/main/org/hornetq/core/server/impl/ServerProducerCreditManager.java 2010-03-22 23:43:26 UTC (rev 8946) +++ trunk/src/main/org/hornetq/core/server/impl/ServerProducerCreditManager.java 2010-03-23 15:11:11 UTC (rev 8947) @@ -1,34 +0,0 @@ -/* - * Copyright 2009 Red Hat, Inc. - * Red Hat licenses this file to you under the Apache License, version - * 2.0 (the "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * http://www.apache.org/licenses/LICENSE-2.0 - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or - * implied. See the License for the specific language governing - * permissions and limitations under the License. - */ - -package org.hornetq.core.server.impl; - -import org.hornetq.core.paging.PagingStore; - -/** - * A ServerProducerCreditManager - * - * @author <a href="mailto:tim.fox@jboss.com">Tim Fox</a> - * - * - */ -public interface ServerProducerCreditManager -{ - int creditsReleased(int credits); - - int acquireCredits(int credits, CreditsAvailableRunnable runnable); - - int waitingEntries(); - - PagingStore getStore(); -} Deleted: trunk/src/main/org/hornetq/core/server/impl/ServerProducerCreditManagerImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/server/impl/ServerProducerCreditManagerImpl.java 2010-03-22 23:43:26 UTC (rev 8946) +++ trunk/src/main/org/hornetq/core/server/impl/ServerProducerCreditManagerImpl.java 2010-03-23 15:11:11 UTC (rev 8947) @@ -1,145 +0,0 @@ -/* - * Copyright 2009 Red Hat, Inc. - * Red Hat licenses this file to you under the Apache License, version - * 2.0 (the "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * http://www.apache.org/licenses/LICENSE-2.0 - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or - * implied. See the License for the specific language governing - * permissions and limitations under the License. - */ - -package org.hornetq.core.server.impl; - -import java.util.Queue; -import java.util.concurrent.ConcurrentLinkedQueue; - -import org.hornetq.core.logging.Logger; -import org.hornetq.core.paging.PagingStore; - -/** - * A ProducerCreditManager - * - * @author <a href="mailto:tim.fox@jboss.com">Tim Fox</a> - * - * - */ -public class ServerProducerCreditManagerImpl implements ServerProducerCreditManager -{ - private static final Logger log = Logger.getLogger(ServerProducerCreditManagerImpl.class); - - private final Queue<WaitingEntry> waiting = new ConcurrentLinkedQueue<WaitingEntry>(); - - private final PagingStore pagingStore; - - public ServerProducerCreditManagerImpl(final PagingStore pagingStore) - { - this.pagingStore = pagingStore; - } - - private static final class WaitingEntry - { - final CreditsAvailableRunnable waiting; - - volatile int credits; - - volatile boolean closed; - - WaitingEntry(final CreditsAvailableRunnable waiting, final int credits) - { - this.waiting = waiting; - - this.credits = credits; - } - - void close() - { - closed = true; - } - } - - public int creditsReleased(int credits) - { - int init = credits; - - while (true) - { - WaitingEntry entry = waiting.peek(); - - if (entry != null) - { - if (entry.credits <= credits) - { - waiting.remove(); - - boolean sent = sendCredits(entry.waiting, entry.credits); - - if (sent) - { - credits -= entry.credits; - - if (entry.credits == credits) - { - break; - } - } - } - else - { - entry.credits -= credits; - - boolean sent = sendCredits(entry.waiting, credits); - - if (sent) - { - credits = 0; - - break; - } - else - { - waiting.remove(); - } - } - } - else - { - break; - } - } - - return init - credits; - } - - public int acquireCredits(final int credits, final CreditsAvailableRunnable runnable) - { - int available = pagingStore.getAvailableProducerCredits(credits); - - if (available < credits) - { - WaitingEntry entry = new WaitingEntry(runnable, credits - available); - - waiting.add(entry); - } - - return available; - } - - public int waitingEntries() - { - return waiting.size(); - } - - private boolean sendCredits(final CreditsAvailableRunnable runnable, final int credits) - { - return runnable.run(credits); - } - - public PagingStore getStore() - { - return pagingStore; - } - -} Modified: trunk/src/main/org/hornetq/core/server/impl/ServerSessionImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/server/impl/ServerSessionImpl.java 2010-03-22 23:43:26 UTC (rev 8946) +++ trunk/src/main/org/hornetq/core/server/impl/ServerSessionImpl.java 2010-03-23 15:11:11 UTC (rev 8947) @@ -83,8 +83,6 @@ // Attributes ---------------------------------------------------------------------------- - // private final long id; - private final String username; private final String password; @@ -130,10 +128,6 @@ // The current currentLargeMessage being processed private volatile LargeServerMessage currentLargeMessage; - private boolean closed; - - private final Map<SimpleString, CreditManagerHolder> creditManagerHolders = new HashMap<SimpleString, CreditManagerHolder>(); - private final RoutingContext routingContext = new RoutingContextImpl(null); private final SessionCallback callback; @@ -274,15 +268,6 @@ remotingConnection.removeFailureListener(this); - // Return any outstanding credits - - closed = true; - - for (CreditManagerHolder holder : creditManagerHolders.values()) - { - holder.store.returnProducerCredits(holder.outstandingCredits); - } - callback.closed(); } @@ -348,7 +333,7 @@ final SimpleString filterString, final boolean temporary, final boolean durable) throws Exception - { + { if (durable) { // make sure the user has privileges to create this queue @@ -378,11 +363,6 @@ if (postOffice.getBinding(name) != null) { postOffice.removeBinding(name); - - if (postOffice.getBindingsForAddress(name).getBindings().size() == 0) - { - creditManagerHolders.remove(name); - } } } catch (Exception e) @@ -406,11 +386,6 @@ server.destroyQueue(name, this); failureRunners.remove(name); - - if (postOffice.getBindingsForAddress(name).getBindings().size() == 0) - { - creditManagerHolders.remove(name); - } } public QueueQueryResult executeQueueQuery(final SimpleString name) throws Exception @@ -929,37 +904,35 @@ currentLargeMessage = msg; } - + + private volatile SimpleString defaultAddress; + public void send(final ServerMessage message) throws Exception { - try + long id = storageManager.generateUniqueID(); + + message.setMessageID(id); + message.encodeMessageIDToBuffer(); + + if (message.getAddress() == null) { - long id = storageManager.generateUniqueID(); + message.setAddress(defaultAddress); + } - message.setMessageID(id); - message.encodeMessageIDToBuffer(); + if (message.getAddress().equals(managementAddress)) + { + // It's a management message - if (message.getAddress().equals(managementAddress)) - { - // It's a management message - - handleManagementMessage(message); - } - else - { - doSend(message); - } + handleManagementMessage(message); } - finally + else { - try - { - releaseOutStanding(message, message.getEncodeSize()); - } - catch (Exception e) - { - ServerSessionImpl.log.error("Failed to release outstanding credits", e); - } + doSend(message); + } + + if (defaultAddress == null) + { + defaultAddress = message.getAddress(); } } @@ -973,8 +946,6 @@ // Immediately release the credits for the continuations- these don't contribute to the in-memory size // of the message - releaseOutStanding(currentLargeMessage, packetSize); - currentLargeMessage.addBytes(body); if (!continues) @@ -983,49 +954,21 @@ doSend(currentLargeMessage); - releaseOutStanding(currentLargeMessage, currentLargeMessage.getEncodeSize()); - currentLargeMessage = null; } } public void requestProducerCredits(final SimpleString address, final int credits) throws Exception - { - final CreditManagerHolder holder = getCreditManagerHolder(address); - - // Requesting -ve credits means returning them - - if (credits < 0) + { + PagingStore store = postOffice.getPagingManager().getPageStore(address); + + store.executeRunnableWhenMemoryAvailable(new Runnable() { - releaseOutStanding(address, -credits); - } - else - { - int gotCredits = holder.manager.acquireCredits(credits, new CreditsAvailableRunnable() + public void run() { - public boolean run(final int credits) - { - synchronized (ServerSessionImpl.this) - { - if (!closed) - { - sendProducerCredits(holder, credits, address); - - return true; - } - else - { - return false; - } - } - } - }); - - if (gotCredits > 0) - { - sendProducerCredits(holder, gotCredits, address); + callback.sendProducerCreditsMessage(credits, address); } - } + }); } public void setTransferring(final boolean transferring) @@ -1184,58 +1127,8 @@ } } - /* - * The way flow producer flow control works is as follows: - * The client can only send messages as long as it has credits. It requests credits from the server - * which the server immediately sends if it has them. If the address is full it adds the request to - * a list of waiting and the credits will get sent as soon as some free up. - * The amount of real memory taken up by a message and references on the server is likely to be larger - * than the encode size of the message. Therefore when a message arrives on the server, more credits - * are taken corresponding to the real in memory size of the message and refs, and the original credits are - * returned. When a session closes any outstanding credits will be returned. - * - */ - private void releaseOutStanding(final ServerMessage message, final int credits) throws Exception - { - releaseOutStanding(message.getAddress(), credits); - } - - private void releaseOutStanding(final SimpleString address, final int credits) throws Exception - { - CreditManagerHolder holder = getCreditManagerHolder(address); - - holder.outstandingCredits -= credits; - - holder.store.returnProducerCredits(credits); - } - - private CreditManagerHolder getCreditManagerHolder(final SimpleString address) throws Exception - { - CreditManagerHolder holder = creditManagerHolders.get(address); - - if (holder == null) - { - PagingStore store = postOffice.getPagingManager().getPageStore(address); - - holder = new CreditManagerHolder(store); - - creditManagerHolders.put(address, holder); - } - - return holder; - } - - private void sendProducerCredits(final CreditManagerHolder holder, final int credits, final SimpleString address) - { - holder.outstandingCredits += credits; - - callback.sendProducerCreditsMessage(credits, address, -1); - } - private void doSend(final ServerMessage msg) throws Exception { - // Look up the paging store - // check the user has write access to this address. try { @@ -1268,19 +1161,4 @@ routingContext.clear(); } - private static final class CreditManagerHolder - { - CreditManagerHolder(final PagingStore store) - { - this.store = store; - - manager = store.getProducerCreditManager(); - } - - final PagingStore store; - - final ServerProducerCreditManager manager; - - volatile int outstandingCredits; - } } Deleted: trunk/src/main/org/hornetq/integration/transports/netty/ChannelPipelineSupport.java =================================================================== --- trunk/src/main/org/hornetq/integration/transports/netty/ChannelPipelineSupport.java 2010-03-22 23:43:26 UTC (rev 8946) +++ trunk/src/main/org/hornetq/integration/transports/netty/ChannelPipelineSupport.java 2010-03-23 15:11:11 UTC (rev 8947) @@ -1,79 +0,0 @@ -/* - * Copyright 2009 Red Hat, Inc. - * Red Hat licenses this file to you under the Apache License, version - * 2.0 (the "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * http://www.apache.org/licenses/LICENSE-2.0 - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or - * implied. See the License for the specific language governing - * permissions and limitations under the License. - */ - -package org.hornetq.integration.transports.netty; - -import javax.net.ssl.SSLContext; -import javax.net.ssl.SSLEngine; - -import org.hornetq.spi.core.protocol.ProtocolType; -import org.hornetq.spi.core.remoting.BufferDecoder; -import org.jboss.netty.channel.ChannelPipeline; -import org.jboss.netty.handler.ssl.SslHandler; - -/** - * @author <a href="mailto:jmesnil@redhat.com">Jeff Mesnil</a> - * @author <a href="mailto:tim.fox@jboss.com">Tim Fox</a> - * @author <a href="mailto:clebert.suconic@jboss.com">Clebert Suconic</a> - * @author <a href="mailto:tlee@redhat.com">Trustin Lee</a> - * @version $Rev$, $Date$ - */ -public class ChannelPipelineSupport -{ - // Constants ----------------------------------------------------- - - // Attributes ---------------------------------------------------- - - // Static -------------------------------------------------------- - - // Constructors -------------------------------------------------- - - private ChannelPipelineSupport() - { - // Unused - } - - // Public -------------------------------------------------------- - - public static void addCodecFilter(final ProtocolType protocol, final ChannelPipeline pipeline, final BufferDecoder decoder) - { - assert pipeline != null; - - if (protocol == ProtocolType.CORE) - { - //Core protocol uses it's own optimised decoder - pipeline.addLast("decoder", new HornetQFrameDecoder2()); - } - else - { - pipeline.addLast("decoder", new HornetQFrameDecoder(decoder)); - } - } - - public static void addSSLFilter(final ChannelPipeline pipeline, final SSLContext context, final boolean client) throws Exception - { - SSLEngine engine = context.createSSLEngine(); - engine.setUseClientMode(client); - if (client) - { - engine.setWantClientAuth(true); - } - - SslHandler handler = new SslHandler(engine); - pipeline.addLast("ssl", handler); - } - - // Package protected --------------------------------------------- - - // Inner classes ------------------------------------------------- -} Modified: trunk/src/main/org/hornetq/integration/transports/netty/HornetQChannelHandler.java =================================================================== --- trunk/src/main/org/hornetq/integration/transports/netty/HornetQChannelHandler.java 2010-03-22 23:43:26 UTC (rev 8946) +++ trunk/src/main/org/hornetq/integration/transports/netty/HornetQChannelHandler.java 2010-03-23 15:11:11 UTC (rev 8947) @@ -58,13 +58,13 @@ group.add(e.getChannel()); ctx.sendUpstream(e); } - + @Override public void messageReceived(final ChannelHandlerContext ctx, final MessageEvent e) throws Exception { ChannelBuffer buffer = (ChannelBuffer)e.getMessage(); - - handler.bufferReceived(e.getChannel().getId(), new ChannelBufferWrapper(buffer)); + + handler.bufferReceived(e.getChannel().getId(), new ChannelBufferWrapper(buffer)); } @Override @@ -113,5 +113,6 @@ HornetQChannelHandler.log.error("failed to notify the listener:", ex); } } - } + } + } Modified: trunk/src/main/org/hornetq/integration/transports/netty/NettyAcceptor.java =================================================================== --- trunk/src/main/org/hornetq/integration/transports/netty/NettyAcceptor.java 2010-03-22 23:43:26 UTC (rev 8946) +++ trunk/src/main/org/hornetq/integration/transports/netty/NettyAcceptor.java 2010-03-23 15:11:11 UTC (rev 8947) @@ -15,7 +15,9 @@ import java.net.InetSocketAddress; import java.net.SocketAddress; +import java.util.ArrayList; import java.util.Iterator; +import java.util.List; import java.util.Map; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; @@ -25,6 +27,7 @@ import java.util.concurrent.TimeUnit; import javax.net.ssl.SSLContext; +import javax.net.ssl.SSLEngine; import org.hornetq.api.core.HornetQException; import org.hornetq.api.core.SimpleString; @@ -48,12 +51,13 @@ import org.jboss.netty.channel.ChannelFactory; import org.jboss.netty.channel.ChannelFuture; import org.jboss.netty.channel.ChannelFutureListener; +import org.jboss.netty.channel.ChannelHandler; import org.jboss.netty.channel.ChannelHandlerContext; import org.jboss.netty.channel.ChannelPipeline; import org.jboss.netty.channel.ChannelPipelineCoverage; import org.jboss.netty.channel.ChannelPipelineFactory; import org.jboss.netty.channel.ChannelStateEvent; -import org.jboss.netty.channel.Channels; +import org.jboss.netty.channel.StaticChannelPipeline; import org.jboss.netty.channel.group.ChannelGroup; import org.jboss.netty.channel.group.ChannelGroupFuture; import org.jboss.netty.channel.group.DefaultChannelGroup; @@ -89,7 +93,7 @@ private ServerBootstrap bootstrap; private final BufferHandler handler; - + private final BufferDecoder decoder; private final ConnectionLifeCycleListener listener; @@ -126,6 +130,10 @@ private final int tcpReceiveBufferSize; + private final int nioRemotingThreads; + + private final int batchingBufferSize; + private final HttpKeepAliveRunnable httpKeepAliveRunnable; private final ConcurrentMap<Object, Connection> connections = new ConcurrentHashMap<Object, Connection>(); @@ -135,7 +143,7 @@ private NotificationService notificationService; private VirtualExecutorService bossExecutor; - + private boolean paused; public NettyAcceptor(final Map<String, Object> configuration, @@ -146,11 +154,11 @@ final ScheduledExecutorService scheduledThreadPool) { this.handler = handler; - + this.decoder = decoder; this.listener = listener; - + sslEnabled = ConfigurationHelper.getBooleanProperty(TransportConstants.SSL_ENABLED_PROP_NAME, TransportConstants.DEFAULT_SSL_ENABLED, configuration); @@ -184,14 +192,22 @@ TransportConstants.DEFAULT_USE_NIO_SERVER, configuration); + nioRemotingThreads = ConfigurationHelper.getIntProperty(TransportConstants.NIO_REMOTING_THREADS_PROPNAME, + -1, + configuration); + + batchingBufferSize = ConfigurationHelper.getIntProperty(TransportConstants.BATCHING_BUFFER_SIZE_PROPNAME, + TransportConstants.DEFAULT_BATCHING_BUFFER_SIZE, + configuration); + useInvm = ConfigurationHelper.getBooleanProperty(TransportConstants.USE_INVM_PROP_NAME, TransportConstants.DEFAULT_USE_INVM, configuration); String protocolStr = ConfigurationHelper.getStringProperty(TransportConstants.PROTOCOL_PROP_NAME, - TransportConstants.DEFAULT_PROTOCOL, - configuration); + TransportConstants.DEFAULT_PROTOCOL, + configuration); protocol = ProtocolType.valueOf(protocolStr.toUpperCase()); - + host = ConfigurationHelper.getStringProperty(TransportConstants.HOST_PROP_NAME, TransportConstants.DEFAULT_HOST, configuration); @@ -251,12 +267,26 @@ } else if (useNio) { - channelFactory = new NioServerSocketChannelFactory(bossExecutor, workerExecutor); + int threadsToUse; + + if (nioRemotingThreads == -1) + { + // Default to number of cores * 3 + + threadsToUse = Runtime.getRuntime().availableProcessors() * 3; + } + else + { + threadsToUse = this.nioRemotingThreads; + } + + channelFactory = new NioServerSocketChannelFactory(bossExecutor, workerExecutor, threadsToUse); } else { channelFactory = new OioServerSocketChannelFactory(bossExecutor, workerExecutor); } + bootstrap = new ServerBootstrap(channelFactory); final SSLContext context; @@ -284,20 +314,42 @@ { public ChannelPipeline getPipeline() throws Exception { - ChannelPipeline pipeline = Channels.pipeline(); + List<ChannelHandler> handlers = new ArrayList<ChannelHandler>(); + if (sslEnabled) { - ChannelPipelineSupport.addSSLFilter(pipeline, context, false); + SSLEngine engine = context.createSSLEngine(); + + engine.setUseClientMode(false); + + SslHandler handler = new SslHandler(engine); + + handlers.add(handler); } + if (httpEnabled) { - pipeline.addLast("httpRequestDecoder", new HttpRequestDecoder()); - pipeline.addLast("httpResponseEncoder", new HttpResponseEncoder()); - pipeline.addLast("httphandler", new HttpAcceptorHandler(httpKeepAliveRunnable, httpResponseTime)); + handlers.add(new HttpRequestDecoder()); + + handlers.add(new HttpResponseEncoder()); + + handlers.add(new HttpAcceptorHandler(httpKeepAliveRunnable, httpResponseTime)); } - ChannelPipelineSupport.addCodecFilter(protocol, pipeline, decoder); - pipeline.addLast("handler", new HornetQServerChannelHandler(channelGroup, handler, new Listener())); + if (protocol == ProtocolType.CORE) + { + // Core protocol uses it's own optimised decoder + handlers.add(new HornetQFrameDecoder2()); + } + else + { + handlers.add(new HornetQFrameDecoder(decoder)); + } + + handlers.add(new HornetQServerChannelHandler(channelGroup, handler, new Listener())); + + ChannelPipeline pipeline = new StaticChannelPipeline(handlers.toArray(new ChannelHandler[handlers.size()])); + return pipeline; } }; @@ -497,7 +549,7 @@ @Override public void channelConnected(final ChannelHandlerContext ctx, final ChannelStateEvent e) throws Exception { - new NettyConnection(e.getChannel(), new Listener()); + new NettyConnection(e.getChannel(), new Listener(), httpEnabled ? -1 : batchingBufferSize); SslHandler sslHandler = ctx.getPipeline().get(SslHandler.class); if (sslHandler != null) Modified: trunk/src/main/org/hornetq/integration/transports/netty/NettyConnection.java =================================================================== --- trunk/src/main/org/hornetq/integration/transports/netty/NettyConnection.java 2010-03-22 23:43:26 UTC (rev 8946) +++ trunk/src/main/org/hornetq/integration/transports/netty/NettyConnection.java 2010-03-23 15:11:11 UTC (rev 8947) @@ -44,16 +44,21 @@ private boolean closed; private final ConnectionLifeCycleListener listener; + + private final int batchingBufferSize; // Static -------------------------------------------------------- // Constructors -------------------------------------------------- - public NettyConnection(final Channel channel, final ConnectionLifeCycleListener listener) + public NettyConnection(final Channel channel, final ConnectionLifeCycleListener listener, + final int batchingBufferSize) { this.channel = channel; this.listener = listener; + + this.batchingBufferSize = batchingBufferSize; listener.connectionCreated(this, ProtocolType.CORE); } @@ -144,6 +149,11 @@ { return channel.getRemoteAddress().toString(); } + + public int getBatchingBufferSize() + { + return batchingBufferSize; + } // Public -------------------------------------------------------- Modified: trunk/src/main/org/hornetq/integration/transports/netty/NettyConnector.java =================================================================== --- trunk/src/main/org/hornetq/integration/transports/netty/NettyConnector.java 2010-03-22 23:43:26 UTC (rev 8946) +++ trunk/src/main/org/hornetq/integration/transports/netty/NettyConnector.java 2010-03-23 15:11:11 UTC (rev 8947) @@ -17,6 +17,8 @@ import java.net.SocketAddress; import java.net.URI; import java.net.URISyntaxException; +import java.util.ArrayList; +import java.util.List; import java.util.Map; import java.util.Set; import java.util.concurrent.ConcurrentHashMap; @@ -26,7 +28,7 @@ import java.util.concurrent.TimeUnit; import javax.net.ssl.SSLContext; -import javax.net.ssl.SSLException; +import javax.net.ssl.SSLEngine; import org.hornetq.api.core.HornetQException; import org.hornetq.core.logging.Logger; @@ -43,6 +45,7 @@ import org.jboss.netty.buffer.ChannelBuffer; import org.jboss.netty.channel.Channel; import org.jboss.netty.channel.ChannelFuture; +import org.jboss.netty.channel.ChannelHandler; import org.jboss.netty.channel.ChannelHandlerContext; import org.jboss.netty.channel.ChannelPipeline; import org.jboss.netty.channel.ChannelPipelineCoverage; @@ -51,6 +54,7 @@ import org.jboss.netty.channel.Channels; import org.jboss.netty.channel.MessageEvent; import org.jboss.netty.channel.SimpleChannelHandler; +import org.jboss.netty.channel.StaticChannelPipeline; import org.jboss.netty.channel.UpstreamMessageEvent; import org.jboss.netty.channel.group.ChannelGroup; import org.jboss.netty.channel.group.DefaultChannelGroup; @@ -129,12 +133,16 @@ private final String servletPath; + private final int nioRemotingThreads; + private final VirtualExecutorService virtualExecutor; private final ScheduledExecutorService scheduledThreadPool; private final Executor closeExecutor; + private final int batchingBufferSize; + // Static -------------------------------------------------------- // Constructors -------------------------------------------------- @@ -193,6 +201,11 @@ useNio = ConfigurationHelper.getBooleanProperty(TransportConstants.USE_NIO_PROP_NAME, TransportConstants.DEFAULT_USE_NIO_CLIENT, configuration); + + nioRemotingThreads = ConfigurationHelper.getIntProperty(TransportConstants.NIO_REMOTING_THREADS_PROPNAME, + -1, + configuration); + useServlet = ConfigurationHelper.getBooleanProperty(TransportConstants.USE_SERVLET_PROP_NAME, TransportConstants.DEFAULT_USE_SERVLET, configuration); @@ -232,6 +245,10 @@ virtualExecutor = new VirtualExecutorService(threadPool); this.scheduledThreadPool = scheduledThreadPool; + + batchingBufferSize = ConfigurationHelper.getIntProperty(TransportConstants.BATCHING_BUFFER_SIZE_PROPNAME, + TransportConstants.DEFAULT_BATCHING_BUFFER_SIZE, + configuration); } public synchronized void start() @@ -243,7 +260,20 @@ if (useNio) { - channelFactory = new NioClientSocketChannelFactory(virtualExecutor, virtualExecutor); + int threadsToUse; + + if (nioRemotingThreads == -1) + { + // Default to number of cores * 3 + + threadsToUse = Runtime.getRuntime().availableProcessors() * 3; + } + else + { + threadsToUse = this.nioRemotingThreads; + } + + channelFactory = new NioClientSocketChannelFactory(virtualExecutor, virtualExecutor, threadsToUse); } else { @@ -300,19 +330,36 @@ { public ChannelPipeline getPipeline() throws Exception { - ChannelPipeline pipeline = Channels.pipeline(); - if (sslEnabled && !useServlet) + List<ChannelHandler> handlers = new ArrayList<ChannelHandler>(); + + if (sslEnabled) { - ChannelPipelineSupport.addSSLFilter(pipeline, context, true); + SSLEngine engine = context.createSSLEngine(); + + engine.setUseClientMode(true); + + engine.setWantClientAuth(true); + + SslHandler handler = new SslHandler(engine); + + handlers.add(handler); } + if (httpEnabled) { - pipeline.addLast("httpRequestEncoder", new HttpRequestEncoder()); - pipeline.addLast("httpResponseDecoder", new HttpResponseDecoder()); - pipeline.addLast("httphandler", new HttpHandler()); + handlers.add(new HttpRequestEncoder()); + + handlers.add(new HttpResponseDecoder()); + + handlers.add(new HttpHandler()); } - ChannelPipelineSupport.addCodecFilter(ProtocolType.CORE, pipeline, null); - pipeline.addLast("handler", new HornetQClientChannelHandler(channelGroup, handler, new Listener())); + + handlers.add(new HornetQFrameDecoder2()); + + handlers.add(new HornetQClientChannelHandler(channelGroup, handler, new Listener())); + + ChannelPipeline pipeline = new StaticChannelPipeline(handlers.toArray(new ChannelHandler[handlers.size()])); + return pipeline; } }); @@ -383,23 +430,15 @@ SslHandler sslHandler = ch.getPipeline().get(SslHandler.class); if (sslHandler != null) { - try + ChannelFuture handshakeFuture = sslHandler.handshake(ch); + handshakeFuture.awaitUninterruptibly(); + if (handshakeFuture.isSuccess()) { - ChannelFuture handshakeFuture = sslHandler.handshake(ch); - handshakeFuture.awaitUninterruptibly(); - if (handshakeFuture.isSuccess()) - { - ch.getPipeline().get(HornetQChannelHandler.class).active = true; - } - else - { - ch.close().awaitUninterruptibly(); - return null; - } + ch.getPipeline().get(HornetQChannelHandler.class).active = true; } - catch (SSLException e) + else { - ch.close(); + ch.close().awaitUninterruptibly(); return null; } } @@ -408,7 +447,7 @@ ch.getPipeline().get(HornetQChannelHandler.class).active = true; } - NettyConnection conn = new NettyConnection(ch, new Listener()); + NettyConnection conn = new NettyConnection(ch, new Listener(), batchingBufferSize); return conn; } Modified: trunk/src/main/org/hornetq/integration/transports/netty/TransportConstants.java =================================================================== --- trunk/src/main/org/hornetq/integration/transports/netty/TransportConstants.java 2010-03-22 23:43:26 UTC (rev 8946) +++ trunk/src/main/org/hornetq/integration/transports/netty/TransportConstants.java 2010-03-23 15:11:11 UTC (rev 8947) @@ -67,10 +67,14 @@ public static final String TCP_SENDBUFFER_SIZE_PROPNAME = "tcp-send-buffer-size"; public static final String TCP_RECEIVEBUFFER_SIZE_PROPNAME = "tcp-receive-buffer-size"; + + public static final String NIO_REMOTING_THREADS_PROPNAME = "nio-remoting-threads"; + + public static final String BATCHING_BUFFER_SIZE_PROPNAME = "batching-buffer-size"; public static final boolean DEFAULT_SSL_ENABLED = false; - public static final boolean DEFAULT_USE_NIO_SERVER = true; + public static final boolean DEFAULT_USE_NIO_SERVER = false; // For client, using old IO can be quicker public static final boolean DEFAULT_USE_NIO_CLIENT = false; @@ -100,6 +104,8 @@ public static final int DEFAULT_TCP_SENDBUFFER_SIZE = 32768; public static final int DEFAULT_TCP_RECEIVEBUFFER_SIZE = 32768; + + public static final int DEFAULT_BATCHING_BUFFER_SIZE = 8192; public static final boolean DEFAULT_HTTP_ENABLED = false; @@ -138,6 +144,8 @@ allowableAcceptorKeys.add(TransportConstants.TCP_NODELAY_PROPNAME); allowableAcceptorKeys.add(TransportConstants.TCP_SENDBUFFER_SIZE_PROPNAME); allowableAcceptorKeys.add(TransportConstants.TCP_RECEIVEBUFFER_SIZE_PROPNAME); + allowableAcceptorKeys.add(TransportConstants.NIO_REMOTING_THREADS_PROPNAME); + allowableAcceptorKeys.add(TransportConstants.BATCHING_BUFFER_SIZE_PROPNAME); ALLOWABLE_ACCEPTOR_KEYS = Collections.unmodifiableSet(allowableAcceptorKeys); @@ -157,6 +165,8 @@ allowableConnectorKeys.add(TransportConstants.TCP_NODELAY_PROPNAME); allowableConnectorKeys.add(TransportConstants.TCP_SENDBUFFER_SIZE_PROPNAME); allowableConnectorKeys.add(TransportConstants.TCP_RECEIVEBUFFER_SIZE_PROPNAME); + allowableConnectorKeys.add(TransportConstants.NIO_REMOTING_THREADS_PROPNAME); + allowableConnectorKeys.add(TransportConstants.BATCHING_BUFFER_SIZE_PROPNAME); ALLOWABLE_CONNECTOR_KEYS = Collections.unmodifiableSet(allowableConnectorKeys); } Modified: trunk/src/main/org/hornetq/jms/client/HornetQMessage.java =================================================================== --- trunk/src/main/org/hornetq/jms/client/HornetQMessage.java 2010-03-22 23:43:26 UTC (rev 8946) +++ trunk/src/main/org/hornetq/jms/client/HornetQMessage.java 2010-03-23 15:11:11 UTC (rev 8947) @@ -42,6 +42,7 @@ import org.hornetq.api.jms.HornetQJMSConstants; import org.hornetq.core.client.impl.ClientMessageImpl; import org.hornetq.core.logging.Logger; +import org.hornetq.utils.UUID; /** * HornetQ implementation of a JMS Message. @@ -69,7 +70,9 @@ private static final SimpleString CORRELATIONID_HEADER_NAME = new SimpleString("JMSCorrelationID"); - public static final SimpleString HORNETQ_MESSAGE_ID = new SimpleString("JMSMessageID"); + //public static final SimpleString HORNETQ_MESSAGE_ID = new SimpleString("_HQI"); + + public static final SimpleString JMSMESSAGEID_HEADER_NAME = new SimpleString("JMSMessageID"); private static final SimpleString TYPE_HEADER_NAME = new SimpleString("JMSType"); @@ -101,8 +104,7 @@ for (Map.Entry<String, Object> entry : coreMessage.entrySet()) { - if (entry.getKey().equals("messageID") || entry.getKey().equals("destination") || - entry.getKey().equals("type") || + if (entry.getKey().equals("type") || entry.getKey().equals("durable") || entry.getKey().equals("expiration") || entry.getKey().equals("timestamp") || @@ -305,17 +307,45 @@ // javax.jmx.Message implementation ------------------------------ +// public String getJMSMessageID() +// { +// if (msgID == null) +// { +// byte[] bytes = message.getBytesProperty(HornetQMessage.HORNETQ_MESSAGE_ID); +// +// msgID = bytes == null ? null : "ID:" + new UUID(UUID.TYPE_TIME_BASED, bytes).toString(); +// } +// return msgID; +// } + public String getJMSMessageID() { if (msgID == null) { - SimpleString id = message.getSimpleStringProperty(HornetQMessage.HORNETQ_MESSAGE_ID); - - msgID = id == null ? null : id.toString(); + msgID = message.getStringProperty(HornetQMessage.JMSMESSAGEID_HEADER_NAME); } return msgID; } +// public void setJMSMessageID(final String jmsMessageID) throws JMSException +// { +// if (jmsMessageID != null && !jmsMessageID.startsWith("ID:")) +// { +// throw new JMSException("JMSMessageID must start with ID:"); +// } +// +// if (jmsMessageID == null) +// { +// message.removeProperty(HornetQMessage.HORNETQ_MESSAGE_ID); +// } +// else +// { +// message.putStringProperty(HornetQMessage.HORNETQ_MESSAGE_ID, new SimpleString(jmsMessageID)); +// } +// +// msgID = jmsMessageID; +// } + public void setJMSMessageID(final String jmsMessageID) throws JMSException { if (jmsMessageID != null && !jmsMessageID.startsWith("ID:")) @@ -325,11 +355,11 @@ if (jmsMessageID == null) { - message.removeProperty(HornetQMessage.HORNETQ_MESSAGE_ID); + message.removeProperty(HornetQMessage.JMSMESSAGEID_HEADER_NAME); } else { - message.putStringProperty(HornetQMessage.HORNETQ_MESSAGE_ID, new SimpleString(jmsMessageID)); + message.putStringProperty(HornetQMessage.JMSMESSAGEID_HEADER_NAME, new SimpleString(jmsMessageID)); } msgID = jmsMessageID; Modified: trunk/src/main/org/hornetq/jms/client/HornetQMessageProducer.java =================================================================== --- trunk/src/main/org/hornetq/jms/client/HornetQMessageProducer.java 2010-03-22 23:43:26 UTC (rev 8946) +++ trunk/src/main/org/hornetq/jms/client/HornetQMessageProducer.java 2010-03-23 15:11:11 UTC (rev 8947) @@ -36,6 +36,7 @@ import org.hornetq.api.core.client.ClientProducer; import org.hornetq.api.core.client.ClientSession; import org.hornetq.core.logging.Logger; +import org.hornetq.utils.UUID; import org.hornetq.utils.UUIDGenerator; /** @@ -76,10 +77,6 @@ private final HornetQDestination defaultDestination; - private final SimpleString messageIDPrefix; - - private volatile long sequenceNumber; - private final ClientSession clientSession; // Constructors -------------------------------------------------- @@ -98,12 +95,6 @@ this.defaultDestination = defaultDestination; this.clientSession = clientSession; - - // TODO the UUID should be generated at the JMS Connection level, - // then session, producers & messages ID could be created using simple sequences - String uuid = UUIDGenerator.getInstance().generateSimpleStringUUID().toString(); - - messageIDPrefix = new SimpleString("ID:" + uuid + ":"); } // MessageProducer implementation -------------------------------- @@ -416,11 +407,13 @@ { // Generate an id - SimpleString msgID = generateMessageID(); - - msg.getCoreMessage().putStringProperty(HornetQMessage.HORNETQ_MESSAGE_ID, msgID); - - msg.resetMessageID(msgID.toString()); + UUID uid = UUIDGenerator.getInstance().generateUUID(); + + //msg.getCoreMessage().putBytesProperty(HornetQMessage.HORNETQ_MESSAGE_ID, uid.asBytes()); + + msg.getCoreMessage().putStringProperty(HornetQMessage.JMSMESSAGEID_HEADER_NAME, new SimpleString("ID:" + uid.toString())); + + msg.resetMessageID(null); } if (foreign) @@ -465,46 +458,6 @@ } } - // This is faster than doing standard String concatenation and conversions from long to string - private SimpleString generateMessageID() - { - byte[] prefixData = messageIDPrefix.getData(); - - int len = prefixData.length + 16 * 2; - - byte[] bytes = new byte[len]; - - System.arraycopy(messageIDPrefix.getData(), 0, bytes, 0, prefixData.length); - - int j = prefixData.length; - - long l = sequenceNumber++; - - for (int i = 0; i < 16; i++) - { - int ch = (int)(l & 0xF); - - l = l >> 4; - - char chr = (char)(ch + 48); - - bytes[j] = (byte)chr; - - j += 2; - } - - return new SimpleString(bytes); - } - - // private SimpleString generateOldMessageID() - // { - // SimpleString ss = new SimpleString(messageIDPrefix.getData()); - // - // ss.concat(String.valueOf(sequenceNumber++)); - // - // return ss; - // } - private void checkClosed() throws JMSException { if (producer.isClosed()) Modified: trunk/src/main/org/hornetq/jms/management/impl/JMSQueueControlImpl.java =================================================================== --- trunk/src/main/org/hornetq/jms/management/impl/JMSQueueControlImpl.java 2010-03-22 23:43:26 UTC (rev 8946) +++ trunk/src/main/org/hornetq/jms/management/impl/JMSQueueControlImpl.java 2010-03-23 15:11:11 UTC (rev 8947) @@ -68,7 +68,7 @@ private static String createFilterForJMSMessageID(final String jmsMessageID) throws Exception { - return HornetQMessage.HORNETQ_MESSAGE_ID + " = '" + jmsMessageID + "'"; + return HornetQMessage.JMSMESSAGEID_HEADER_NAME + " = '" + jmsMessageID + "'"; } static String toJSON(final Map<String, Object>[] messages) @@ -119,7 +119,7 @@ return coreQueueControl.getMessageCount(); } - public int getMessagesAdded() + public long getMessagesAdded() { return coreQueueControl.getMessagesAdded(); } Modified: trunk/src/main/org/hornetq/spi/core/protocol/SessionCallback.java =================================================================== --- trunk/src/main/org/hornetq/spi/core/protocol/SessionCallback.java 2010-03-22 23:43:26 UTC (rev 8946) +++ trunk/src/main/org/hornetq/spi/core/protocol/SessionCallback.java 2010-03-23 15:11:11 UTC (rev 8947) @@ -25,7 +25,7 @@ */ public interface SessionCallback { - void sendProducerCreditsMessage(int credits, SimpleString address, int offset); + void sendProducerCreditsMessage(int credits, SimpleString address); int sendMessage(ServerMessage message, long consumerID, int deliveryCount); Modified: trunk/src/main/org/hornetq/spi/core/remoting/Connection.java =================================================================== --- trunk/src/main/org/hornetq/spi/core/remoting/Connection.java 2010-03-22 23:43:26 UTC (rev 8946) +++ trunk/src/main/org/hornetq/spi/core/remoting/Connection.java 2010-03-23 15:11:11 UTC (rev 8947) @@ -65,4 +65,12 @@ * @return the remote address */ String getRemoteAddress(); + + /** + * The batch size in bytes of the buffer for batching sends + * or -1 if the connection does not support batching + * + * @return + */ + int getBatchingBufferSize(); } \ No newline at end of file Added: trunk/src/main/org/hornetq/utils/ConcurrentHQDeque.java =================================================================== --- trunk/src/main/org/hornetq/utils/ConcurrentHQDeque.java (rev 0) +++ trunk/src/main/org/hornetq/utils/ConcurrentHQDeque.java 2010-03-23 15:11:11 UTC (rev 8947) @@ -0,0 +1,140 @@ +/* + * Copyright 2010 Red Hat, Inc. + * Red Hat licenses this file to you under the Apache License, version + * 2.0 (the "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * http://www.apache.org/licenses/LICENSE-2.0 + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or + * implied. See the License for the specific language governing + * permissions and limitations under the License. + */ + +package org.hornetq.utils; + +import java.util.Iterator; + +import org.hornetq.core.logging.Logger; +import org.hornetq.utils.concurrent.BlockingDeque; +import org.hornetq.utils.concurrent.HQIterator; +import org.hornetq.utils.concurrent.HornetQConcurrentLinkedQueue; +import org.hornetq.utils.concurrent.LinkedBlockingDeque; + +/** + * A ConcurrentHQDeque + * + * @author Tim Fox + * + * + */ +public class ConcurrentHQDeque<T> implements HQDeque<T> +{ + private static final Logger log = Logger.getLogger(ConcurrentHQDeque.class); + + private final HornetQConcurrentLinkedQueue<T> bodyQueue; + + private final BlockingDeque<T> headQueue; + + public ConcurrentHQDeque() + { + this.bodyQueue = new HornetQConcurrentLinkedQueue<T>(); + + this.headQueue = new LinkedBlockingDeque<T>(); + } + + public synchronized void addFirst(T t) + { + headQueue.addFirst(t); + } + + public void addLast(T t) + { + bodyQueue.add(t); + } + + public void clear() + { + bodyQueue.clear(); + headQueue.clear(); + } + + public synchronized T getFirst() + { + if (headQueue.isEmpty()) + { + return bodyQueue.peek(); + } + else + { + return headQueue.peek(); + } + } + + public boolean isEmpty() + { + return bodyQueue.isEmpty() && headQueue.isEmpty(); + } + + public HQIterator<T> iterator() + { + return new Iter(); + } + + public T removeFirst() + { + if (headQueue.isEmpty()) + { + return bodyQueue.remove(); + } + else + { + return headQueue.remove(); + } + } + + private class Iter implements HQIterator<T> + { + private Iterator<T> headIter; + + private HQIterator<T> bodyIter; + + private boolean inHead; + + private Iter() + { + headIter = headQueue.iterator(); + + bodyIter = bodyQueue.hqIterator(); + } + + public T next() + { + if (headIter.hasNext()) + { + inHead = true; + + return headIter.next(); + } + else + { + inHead = false; + + return bodyIter.next(); + } + } + + public void remove() + { + if (inHead) + { + headIter.remove(); + } + else + { + bodyIter.remove(); + } + } + + } +} Modified: trunk/src/main/org/hornetq/utils/HornetQThreadFactory.java =================================================================== --- trunk/src/main/org/hornetq/utils/HornetQThreadFactory.java 2010-03-22 23:43:26 UTC (rev 8946) +++ trunk/src/main/org/hornetq/utils/HornetQThreadFactory.java 2010-03-23 15:11:11 UTC (rev 8947) @@ -15,6 +15,8 @@ import java.util.concurrent.ThreadFactory; import java.util.concurrent.atomic.AtomicInteger; +import org.hornetq.core.logging.Logger; + /** * * A HornetQThreadFactory @@ -24,6 +26,8 @@ */ public class HornetQThreadFactory implements ThreadFactory { + private static final Logger log = Logger.getLogger(HornetQThreadFactory.class); + private final ThreadGroup group; private final AtomicInteger threadCount = new AtomicInteger(0); @@ -59,7 +63,7 @@ { t = new Thread(command, "Thread-" + threadCount.getAndIncrement()); } - + t.setDaemon(daemon); t.setPriority(threadPriority); return t; Copied: trunk/src/main/org/hornetq/utils/PriorityLinkedList.java (from rev 8917, trunk/src/main/org/hornetq/core/list/PriorityLinkedList.java) =================================================================== --- trunk/src/main/org/hornetq/utils/PriorityLinkedList.java (rev 0) +++ trunk/src/main/org/hornetq/utils/PriorityLinkedList.java 2010-03-23 15:11:11 UTC (rev 8947) @@ -0,0 +1,44 @@ +/* + * Copyright 2009 Red Hat, Inc. + * Red Hat licenses this file to you under the Apache License, version + * 2.0 (the "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * http://www.apache.org/licenses/LICENSE-2.0 + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or + * implied. See the License for the specific language governing + * permissions and limitations under the License. + */ + +package org.hornetq.utils; + +import org.hornetq.utils.concurrent.HQIterator; + +/** + * A type of linked list which maintains items according to a priority + * and allows adding and removing of elements at both ends, and peeking + * + * @author <a href="mailto:tim.fox@jboss.com>Tim Fox</a> + * @version <tt>$Revision: 1174 $</tt> + * + * $Id: PrioritizedDeque.java 1174 2006-08-02 14:14:32Z timfox $ + */ +public interface PriorityLinkedList<T> +{ + int addFirst(T t, int priority); + + int addLast(T t, int priority); + + T removeFirst(); + + T peekFirst(); + + void clear(); + + int size(); + + HQIterator<T> iterator(); + + boolean isEmpty(); +} Copied: trunk/src/main/org/hornetq/utils/PriorityLinkedListImpl.java (from rev 8917, trunk/src/main/org/hornetq/core/list/impl/PriorityLinkedListImpl.java) =================================================================== --- trunk/src/main/org/hornetq/utils/PriorityLinkedListImpl.java (rev 0) +++ trunk/src/main/org/hornetq/utils/PriorityLinkedListImpl.java 2010-03-23 15:11:11 UTC (rev 8947) @@ -0,0 +1,211 @@ +/* + * Copyright 2009 Red Hat, Inc. + * Red Hat licenses this file to you under the Apache License, version + * 2.0 (the "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * http://www.apache.org/licenses/LICENSE-2.0 + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or + * implied. See the License for the specific language governing + * permissions and limitations under the License. + */ + +package org.hornetq.utils; + +import java.lang.reflect.Array; +import java.util.NoSuchElementException; +import java.util.concurrent.atomic.AtomicInteger; + +import org.hornetq.core.logging.Logger; +import org.hornetq.utils.concurrent.HQIterator; + +/** + * A priority linked list implementation + * + * It implements this by maintaining an individual LinkedBlockingDeque for each priority level. + * + * @author <a href="mailto:tim.fox@jboss.com>Tim Fox</a> + * @author <a href="mailto:jmesnil@redhat.com>Jeff Mesnil</a> + * @version <tt>$Revision: 1174 $</tt> + * + * $Id: BasicPrioritizedDeque.java 1174 2006-08-02 14:14:32Z timfox $ + */ +public class PriorityLinkedListImpl<T> implements PriorityLinkedList<T> +{ + private static final Logger log = Logger.getLogger(PriorityLinkedListImpl.class); + + private HQDeque<T>[] levels; + + private final int priorities; + + private final AtomicInteger size = new AtomicInteger(0); + + public PriorityLinkedListImpl(final boolean concurrent, final int priorities) + { + this.priorities = priorities; + + levels = (HQDeque<T>[])Array.newInstance(HQDeque.class, priorities); + + for (int i = 0; i < priorities; i++) + { + if (concurrent) + { + levels[i] = new ConcurrentHQDeque<T>(); + } + else + { + levels[i] = new NonConcurrentHQDeque<T>(); + } + } + } + + public int addFirst(final T t, final int priority) + { + int refs = size.incrementAndGet(); + + levels[priority].addFirst(t); + + return refs; + } + + public int addLast(final T t, final int priority) + { + int refs = size.incrementAndGet(); + + levels[priority].addLast(t); + + return refs; + } + + public T removeFirst() + { + T t = null; + + // Initially we are just using a simple prioritization algorithm: + // Highest priority refs always get returned first. + // This could cause starvation of lower priority refs. + + // TODO - A better prioritization algorithm + + for (int i = priorities - 1; i >= 0; i--) + { + HQDeque<T> ll = levels[i]; + + if (!ll.isEmpty()) + { + t = ll.removeFirst(); + break; + } + } + + if (t != null) + { + size.decrementAndGet(); + } + + return t; + } + + public T peekFirst() + { + T t = null; + + for (int i = priorities - 1; i >= 0; i--) + { + HQDeque<T> ll = levels[i]; + if (!ll.isEmpty()) + { + t = ll.getFirst(); + } + if (t != null) + { + break; + } + } + + return t; + } + + public void clear() + { + for (HQDeque<T> list : levels) + { + list.clear(); + } + + size.set(0); + } + + public int size() + { + return size.get(); + } + + public boolean isEmpty() + { + return size.get() == 0; + } + + public HQIterator<T> iterator() + { + return new PriorityLinkedListIterator(); + } + + private class PriorityLinkedListIterator implements HQIterator<T> + { + private int index; + + private HQIterator<T>[] cachedIters = new HQIterator[levels.length]; + + PriorityLinkedListIterator() + { + index = levels.length - 1; + } + + public T next() + { + while (index >= 0) + { + HQIterator<T> iter = cachedIters[index]; + + if (iter == null) + { + iter = cachedIters[index] = levels[index].iterator(); + } + + T t = iter.next(); + + if (t != null) + { + return t; + } + + index--; + + if (index < 0) + { + index = levels.length - 1; + + break; + } + } + + return null; + } + + public void remove() + { + HQIterator<T> iter = cachedIters[index]; + + if (iter == null) + { + throw new NoSuchElementException(); + } + + iter.remove(); + + size.decrementAndGet(); + } + } +} Modified: trunk/src/main/org/hornetq/utils/TypedProperties.java =================================================================== --- trunk/src/main/org/hornetq/utils/TypedProperties.java 2010-03-22 23:43:26 UTC (rev 8946) +++ trunk/src/main/org/hornetq/utils/TypedProperties.java 2010-03-23 15:11:11 UTC (rev 8947) @@ -422,25 +422,26 @@ public boolean containsProperty(final SimpleString key) { - if (properties != null) + if (size == 0) { - return properties.containsKey(key); + return false; + } else - { - return false; + { + return properties.containsKey(key); } } public Set<SimpleString> getPropertyNames() { - if (properties != null) + if (size == 0) { - return properties.keySet(); + return Collections.EMPTY_SET; } else { - return Collections.EMPTY_SET; + return properties.keySet(); } } @@ -455,7 +456,7 @@ else { int numHeaders = buffer.readInt(); - + properties = new HashMap<SimpleString, PropertyValue>(numHeaders); size = 0; @@ -643,7 +644,7 @@ private synchronized Object doGetProperty(final Object key) { - if (properties == null) + if (size == 0) { return null; } Added: trunk/src/main/org/hornetq/utils/concurrent/HQIterator.java =================================================================== --- trunk/src/main/org/hornetq/utils/concurrent/HQIterator.java (rev 0) +++ trunk/src/main/org/hornetq/utils/concurrent/HQIterator.java 2010-03-23 15:11:11 UTC (rev 8947) @@ -0,0 +1,28 @@ +/* + * Copyright 2010 Red Hat, Inc. + * Red Hat licenses this file to you under the Apache License, version + * 2.0 (the "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * http://www.apache.org/licenses/LICENSE-2.0 + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or + * implied. See the License for the specific language governing + * permissions and limitations under the License. + */ + +package org.hornetq.utils.concurrent; + +/** + * A HQIterator + * + * @author Tim Fox + * + * + */ +public interface HQIterator<E> +{ + E next(); + + void remove(); +} Added: trunk/src/main/org/hornetq/utils/concurrent/HornetQConcurrentLinkedQueue.java =================================================================== --- trunk/src/main/org/hornetq/utils/concurrent/HornetQConcurrentLinkedQueue.java (rev 0) +++ trunk/src/main/org/hornetq/utils/concurrent/HornetQConcurrentLinkedQueue.java 2010-03-23 15:11:11 UTC (rev 8947) @@ -0,0 +1,867 @@ +package org.hornetq.utils.concurrent; + +import java.util.AbstractQueue; +import java.util.ArrayList; +import java.util.Collection; +import java.util.Iterator; +import java.util.NoSuchElementException; +import java.util.Queue; +import java.util.concurrent.atomic.AtomicReferenceFieldUpdater; + +import org.hornetq.core.logging.Logger; + +/** + * An unbounded thread-safe {@linkplain Queue queue} based on linked nodes. + * This queue orders elements FIFO (first-in-first-out). + * The <em>head</em> of the queue is that element that has been on the + * queue the longest time. + * The <em>tail</em> of the queue is that element that has been on the + * queue the shortest time. New elements + * are inserted at the tail of the queue, and the queue retrieval + * operations obtain elements at the head of the queue. + * A {@code ConcurrentLinkedQueue} is an appropriate choice when + * many threads will share access to a common collection. + * This queue does not permit {@code null} elements. + * + * <p>This implementation employs an efficient &quot;wait-free&quot; + * algorithm based on one described in <a + * href="http://www.cs.rochester.edu/u/michael/PODC96.html"> Simple, + * Fast, and Practical Non-Blocking and Blocking Concurrent Queue + * Algorithms</a> by Maged M. Michael and Michael L. Scott. + * + * <p>Beware that, unlike in most collections, the {@code size} method + * is <em>NOT</em> a constant-time operation. Because of the + * asynchronous nature of these queues, determining the current number + * of elements requires a traversal of the elements. + * + * <p>This class and its iterator implement all of the + * <em>optional</em> methods of the {@link Collection} and {@link + * Iterator} interfaces. + * + * <p>Memory consistency effects: As with other concurrent + * collections, actions in a thread prior to placing an object into a + * {@code ConcurrentLinkedQueue} + * <a href="package-summary.html#MemoryVisibility"><i>happen-before</i></a> + * actions subsequent to the access or removal of that element from + * the {@code ConcurrentLinkedQueue} in another thread. + * + * <p>This class is a member of the + * <a href="{@docRoot}/../technotes/guides/collections/index.html"> + * Java Collections Framework</a>. + * + * @since 1.5 + * @author Doug Lea + * @param <E> the type of elements held in this collection + * + * Based on the public domain version by Doug Lea which contained this original header: + * + * + * Written by Doug Lea with assistance from members of JCP JSR-166 + * Expert Group and released to the public domain, as explained at + * http://creativecommons.org/licenses/publicdomain + * + * + * Any subsequent modifications done under the terms of the ASL 2: + * + * We use this class rather than the JDK ConcurrentLinkedQueue since we need iterators which will also + * see any nodes added to the queue after the iterator has already iterated to the tail of the queue + * The JDK ConcurrentLinkedQueue does not see any subsequently added + * + * You may obtain a copy of the License at + * http://www.apache.org/licenses/LICENSE-2.0 + * + */ +public class HornetQConcurrentLinkedQueue<E> extends AbstractQueue<E> implements Queue<E>, java.io.Serializable +{ + private static final long serialVersionUID = -1513782900644400377L; + + private static final Logger log = Logger.getLogger(HornetQConcurrentLinkedQueue.class); + + /* + * This is a modification of the Michael & Scott algorithm, + * adapted for a garbage-collected environment, with support for + * interior node deletion (to support remove(Object)). For + * explanation, read the paper. + * + * Note that like most non-blocking algorithms in this package, + * this implementation relies on the fact that in garbage + * collected systems, there is no possibility of ABA problems due + * to recycled nodes, so there is no need to use "counted + * pointers" or related techniques seen in versions used in + * non-GC'ed settings. + * + * The fundamental invariants are: + * - There is exactly one (last) Node with a null next reference, + * which is CASed when enqueueing. This last Node can be + * reached in O(1) time from tail, but tail is merely an + * optimization - it can always be reached in O(N) time from + * head as well. + * - The elements contained in the queue are the non-null items in + * Nodes that are reachable from head. CASing the item + * reference of a Node to null atomically removes it from the + * queue. Reachability of all elements from head must remain + * true even in the case of concurrent modifications that cause + * head to advance. A dequeued Node may remain in use + * indefinitely due to creation of an Iterator or simply a + * poll() that has lost its time slice. + * + * The above might appear to imply that all Nodes are GC-reachable + * from a predecessor dequeued Node. That would cause two problems: + * - allow a rogue Iterator to cause unbounded memory retention + * - cause cross-generational linking of old Nodes to new Nodes if + * a Node was tenured while live, which generational GCs have a + * hard time dealing with, causing repeated major collections. + * However, only non-deleted Nodes need to be reachable from + * dequeued Nodes, and reachability does not necessarily have to + * be of the kind understood by the GC. We use the trick of + * linking a Node that has just been dequeued to itself. Such a + * self-link implicitly means to advance to head. + * + * Both head and tail are permitted to lag. In fact, failing to + * update them every time one could is a significant optimization + * (fewer CASes). This is controlled by local "hops" variables + * that only trigger helping-CASes after experiencing multiple + * lags. + * + * Since head and tail are updated concurrently and independently, + * it is possible for tail to lag behind head (why not)? + * + * CASing a Node's item reference to null atomically removes the + * element from the queue. Iterators skip over Nodes with null + * items. Prior implementations of this class had a race between + * poll() and remove(Object) where the same element would appear + * to be successfully removed by two concurrent operations. The + * method remove(Object) also lazily unlinks deleted Nodes, but + * this is merely an optimization. + * + * When constructing a Node (before enqueuing it) we avoid paying + * for a volatile write to item by using lazySet instead of a + * normal write. This allows the cost of enqueue to be + * "one-and-a-half" CASes. + * + * Both head and tail may or may not point to a Node with a + * non-null item. If the queue is empty, all items must of course + * be null. Upon creation, both head and tail refer to a dummy + * Node with null item. Both head and tail are only updated using + * CAS, so they never regress, although again this is merely an + * optimization. + */ + + private static class Node<E> + { + private volatile E item; + + private volatile Node<E> next; + + private static final AtomicReferenceFieldUpdater<Node, Object> itemFieldUpdater = AtomicReferenceFieldUpdater.newUpdater(Node.class, + Object.class, + "item"); + + private static final AtomicReferenceFieldUpdater<Node, Node> nextFieldUpdater = AtomicReferenceFieldUpdater.newUpdater(Node.class, + Node.class, + "next"); + + Node(E item) + { + // Piggyback on imminent casNext() + lazySetItem(item); + } + + E getItem() + { + return item; + } + + boolean casItem(E cmp, E val) + { + return itemFieldUpdater.compareAndSet(this, cmp, val); + } + + void setItem(E val) + { + item = val; + } + + void lazySetItem(E val) + { + itemFieldUpdater.lazySet(this, val); + } + + void lazySetNext(Node<E> val) + { + + nextFieldUpdater.lazySet(this, val); + } + + Node<E> getNext() + { + return next; + } + + boolean casNext(Node<E> cmp, Node<E> val) + { + return nextFieldUpdater.compareAndSet(this, cmp, val); + } + } + + /** + * A node from which the first live (non-deleted) node (if any) + * can be reached in O(1) time. + * Invariants: + * - all live nodes are reachable from head via succ() + * - head != null + * - (tmp = head).next != tmp || tmp != head + * Non-invariants: + * - head.item may or may not be null. + * - it is permitted for tail to lag behind head, that is, for tail + * to not be reachable from head! + */ + private transient volatile Node<E> head = new Node<E>(null); + + /** + * A node from which the last node on list (that is, the unique + * node with node.next == null) can be reached in O(1) time. + * Invariants: + * - the last node is always reachable from tail via succ() + * - tail != null + * Non-invariants: + * - tail.item may or may not be null. + * - it is permitted for tail to lag behind head, that is, for tail + * to not be reachable from head! + * - tail.next may or may not be self-pointing to tail. + */ + private transient volatile Node<E> tail = head; + + /** + * Creates a {@code ConcurrentLinkedQueue} that is initially empty. + */ + public HornetQConcurrentLinkedQueue() + { + } + + /** + * Creates a {@code ConcurrentLinkedQueue} + * initially containing the elements of the given collection, + * added in traversal order of the collection's iterator. + * @param c the collection of elements to initially contain + * @throws NullPointerException if the specified collection or any + * of its elements are null + */ + public HornetQConcurrentLinkedQueue(Collection<? extends E> c) + { + for (E e : c) + add(e); + } + + // Have to override just to update the javadoc + + /** + * Inserts the specified element at the tail of this queue. + * + * @return {@code true} (as specified by {@link Collection#add}) + * @throws NullPointerException if the specified element is null + */ + public boolean add(E e) + { + return offer(e); + } + + /** + * We don't bother to update head or tail pointers if fewer than + * HOPS links from "true" location. We assume that volatile + * writes are significantly more expensive than volatile reads. + */ + private static final int HOPS = 1; + + /** + * Try to CAS head to p. If successful, repoint old head to itself + * as sentinel for succ(), below. + */ + final void updateHead(Node<E> h, Node<E> p) + { + if (h != p && casHead(h, p)) + h.lazySetNext(h); + } + + /** + * Returns the successor of p, or the head node if p.next has been + * linked to self, which will only be true if traversing with a + * stale pointer that is now off the list. + */ + final Node<E> succ(Node<E> p) + { + Node<E> next = p.getNext(); + return (p == next) ? head : next; + } + + /** + * Inserts the specified element at the tail of this queue. + * + * @return {@code true} (as specified by {@link Queue#offer}) + * @throws NullPointerException if the specified element is null + */ + public boolean offer(E e) + { + if (e == null) + throw new NullPointerException(); + Node<E> n = new Node<E>(e); + retry: for (;;) + { + Node<E> t = tail; + Node<E> p = t; + for (int hops = 0;; hops++) + { + Node<E> next = succ(p); + if (next != null) + { + if (hops > HOPS && t != tail) + continue retry; + p = next; + } + else if (p.casNext(null, n)) + { + if (hops >= HOPS) + casTail(t, n); // Failure is OK. + return true; + } + else + { + p = succ(p); + } + } + } + } + + public E poll() + { + Node<E> h = head; + Node<E> p = h; + for (int hops = 0;; hops++) + { + E item = p.getItem(); + + if (item != null && p.casItem(item, null)) + { + if (hops >= HOPS) + { + Node<E> q = p.getNext(); + updateHead(h, (q != null) ? q : p); + } + return item; + } + Node<E> next = succ(p); + if (next == null) + { + updateHead(h, p); + break; + } + p = next; + } + return null; + } + + public E peek() + { + Node<E> h = head; + Node<E> p = h; + E item; + for (;;) + { + item = p.getItem(); + if (item != null) + break; + Node<E> next = succ(p); + if (next == null) + { + break; + } + p = next; + } + updateHead(h, p); + return item; + } + + /** + * Returns the first live (non-deleted) node on list, or null if none. + * This is yet another variant of poll/peek; here returning the + * first node, not element. We could make peek() a wrapper around + * first(), but that would cost an extra volatile read of item, + * and the need to add a retry loop to deal with the possibility + * of losing a race to a concurrent poll(). + */ + Node<E> first() + { + Node<E> h = head; + Node<E> p = h; + Node<E> result; + for (;;) + { + E item = p.getItem(); + if (item != null) + { + result = p; + break; + } + Node<E> next = succ(p); + if (next == null) + { + result = null; + break; + } + p = next; + } + updateHead(h, p); + return result; + } + + /** + * Returns {@code true} if this queue contains no elements. + * + * @return {@code true} if this queue contains no elements + */ + public boolean isEmpty() + { + return first() == null; + } + + /** + * Returns the number of elements in this queue. If this queue + * contains more than {@code Integer.MAX_VALUE} elements, returns + * {@code Integer.MAX_VALUE}. + * + * <p>Beware that, unlike in most collections, this method is + * <em>NOT</em> a constant-time operation. Because of the + * asynchronous nature of these queues, determining the current + * number of elements requires an O(n) traversal. + * + * @return the number of elements in this queue + */ + public int size() + { + int count = 0; + for (Node<E> p = first(); p != null; p = succ(p)) + { + if (p.getItem() != null) + { + // Collections.size() spec says to max out + if (++count == Integer.MAX_VALUE) + break; + } + } + return count; + } + + /** + * Returns {@code true} if this queue contains the specified element. + * More formally, returns {@code true} if and only if this queue contains + * at least one element {@code e} such that {@code o.equals(e)}. + * + * @param o object to be checked for containment in this queue + * @return {@code true} if this queue contains the specified element + */ + public boolean contains(Object o) + { + if (o == null) + return false; + for (Node<E> p = first(); p != null; p = succ(p)) + { + E item = p.getItem(); + if (item != null && o.equals(item)) + return true; + } + return false; + } + + /** + * Removes a single instance of the specified element from this queue, + * if it is present. More formally, removes an element {@code e} such + * that {@code o.equals(e)}, if this queue contains one or more such + * elements. + * Returns {@code true} if this queue contained the specified element + * (or equivalently, if this queue changed as a result of the call). + * + * @param o element to be removed from this queue, if present + * @return {@code true} if this queue changed as a result of the call + */ + public boolean remove(Object o) + { + if (o == null) + return false; + Node<E> pred = null; + for (Node<E> p = first(); p != null; p = succ(p)) + { + E item = p.getItem(); + if (item != null && o.equals(item) && p.casItem(item, null)) + { + Node<E> next = succ(p); + if (pred != null && next != null) + pred.casNext(p, next); + return true; + } + pred = p; + } + return false; + } + + /** + * Returns an array containing all of the elements in this queue, in + * proper sequence. + * + * <p>The returned array will be "safe" in that no references to it are + * maintained by this queue. (In other words, this method must allocate + * a new array). The caller is thus free to modify the returned array. + * + * <p>This method acts as bridge between array-based and collection-based + * APIs. + * + * @return an array containing all of the elements in this queue + */ + public Object[] toArray() + { + // Use ArrayList to deal with resizing. + ArrayList<E> al = new ArrayList<E>(); + for (Node<E> p = first(); p != null; p = succ(p)) + { + E item = p.getItem(); + if (item != null) + al.add(item); + } + return al.toArray(); + } + + /** + * Returns an array containing all of the elements in this queue, in + * proper sequence; the runtime type of the returned array is that of + * the specified array. If the queue fits in the specified array, it + * is returned therein. Otherwise, a new array is allocated with the + * runtime type of the specified array and the size of this queue. + * + * <p>If this queue fits in the specified array with room to spare + * (i.e., the array has more elements than this queue), the element in + * the array immediately following the end of the queue is set to + * {@code null}. + * + * <p>Like the {@link #toArray()} method, this method acts as bridge between + * array-based and collection-based APIs. Further, this method allows + * precise control over the runtime type of the output array, and may, + * under certain circumstances, be used to save allocation costs. + * + * <p>Suppose {@code x} is a queue known to contain only strings. + * The following code can be used to dump the queue into a newly + * allocated array of {@code String}: + * + * <pre> + * String[] y = x.toArray(new String[0]);</pre> + * + * Note that {@code toArray(new Object[0])} is identical in function to + * {@code toArray()}. + * + * @param a the array into which the elements of the queue are to + * be stored, if it is big enough; otherwise, a new array of the + * same runtime type is allocated for this purpose + * @return an array containing all of the elements in this queue + * @throws ArrayStoreException if the runtime type of the specified array + * is not a supertype of the runtime type of every element in + * this queue + * @throws NullPointerException if the specified array is null + */ + @SuppressWarnings("unchecked") + public <T> T[] toArray(T[] a) + { + // try to use sent-in array + int k = 0; + Node<E> p; + for (p = first(); p != null && k < a.length; p = succ(p)) + { + E item = p.getItem(); + if (item != null) + a[k++] = (T)item; + } + if (p == null) + { + if (k < a.length) + a[k] = null; + return a; + } + + // If won't fit, use ArrayList version + ArrayList<E> al = new ArrayList<E>(); + for (Node<E> q = first(); q != null; q = succ(q)) + { + E item = q.getItem(); + if (item != null) + al.add(item); + } + return al.toArray(a); + } + + /** + * Returns an iterator over the elements in this queue in proper sequence. + * The returned iterator is a "weakly consistent" iterator that + * will never throw {@link java.util.ConcurrentModificationException + * ConcurrentModificationException}, + * and guarantees to traverse elements as they existed upon + * construction of the iterator, and may (but is not guaranteed to) + * reflect any modifications subsequent to construction. + * + * @return an iterator over the elements in this queue in proper sequence + */ + public Iterator<E> iterator() + { + return new Itr(); + } + + private class Itr implements Iterator<E> + { + /** + * Next node to return item for. + */ + private Node<E> nextNode; + + /** + * nextItem holds on to item fields because once we claim + * that an element exists in hasNext(), we must return it in + * the following next() call even if it was in the process of + * being removed when hasNext() was called. + */ + private E nextItem; + + /** + * Node of the last returned item, to support remove. + */ + private Node<E> lastRet; + + Itr() + { + advance(); + } + + /** + * Moves to next valid node and returns item to return for + * next(), or null if no such. + */ + private E advance() + { + lastRet = nextNode; + E x = nextItem; + + Node<E> pred, p; + if (nextNode == null) + { + p = first(); + pred = null; + } + else + { + pred = nextNode; + p = succ(nextNode); + } + + for (;;) + { + if (p == null) + { + nextNode = null; + nextItem = null; + return x; + } + E item = p.getItem(); + if (item != null) + { + nextNode = p; + nextItem = item; + return x; + } + else + { + // skip over nulls + Node<E> next = succ(p); + if (pred != null && next != null) + pred.casNext(p, next); + p = next; + } + } + } + + public boolean hasNext() + { + if (nextNode != null) + { + return true; + } + else + { + // Maybe a new node has been added to the tail since we last computed the nextNode + + if (lastRet.next != null) + { + // We need to compute advance() again + + nextNode = lastRet; + nextItem = lastRet.item; + + advance(); + + return nextNode != null; + } + else + { + return false; + } + } + } + + public E next() + { + if (nextNode == null) + throw new NoSuchElementException(); + return advance(); + } + + public void remove() + { + Node<E> l = lastRet; + if (l == null) + throw new IllegalStateException(); + // rely on a future traversal to relink. + l.setItem(null); + lastRet = null; + } + } + + public HQIterator<E> hqIterator() + { + return new HQIteratorImpl(); + } + + /* + * This "iterator" combines hasNext() and next() into a single method + * If there is no element then next() returns null + * + * With the other iterator if hasNext() returns true, then the next element is cached, so if it + * is deleted by another iterator, the first iterator will still return it even though it has been + * deleted which does not work well for HornetQ + */ + private class HQIteratorImpl implements HQIterator<E> + { + Node<E> currNode; + + public E next() + { + Node<E> pred, p; + if (currNode == null) + { + p = first(); + pred = null; + } + else + { + pred = currNode; + p = succ(currNode); + } + + for (;;) + { + if (p == null) + { + return null; + } + E item = p.getItem(); + if (item != null) + { + currNode = p; + return item; + } + else + { + // skip over nulls + Node<E> next = succ(p); + if (pred != null && next != null) + pred.casNext(p, next); + p = next; + } + } + } + + public void remove() + { + Node<E> l = currNode; + if (l == null) + throw new IllegalStateException(); + // rely on a future traversal to relink. + l.setItem(null); + } + } + + /** + * Save the state to a stream (that is, serialize it). + * + * @serialData All of the elements (each an {@code E}) in + * the proper order, followed by a null + * @param s the stream + */ + private void writeObject(java.io.ObjectOutputStream s) throws java.io.IOException + { + + // Write out any hidden stuff + s.defaultWriteObject(); + + // Write out all elements in the proper order. + for (Node<E> p = first(); p != null; p = succ(p)) + { + Object item = p.getItem(); + if (item != null) + s.writeObject(item); + } + + // Use trailing null as sentinel + s.writeObject(null); + } + + /** + * Reconstitute the Queue instance from a stream (that is, + * deserialize it). + * @param s the stream + */ + private void readObject(java.io.ObjectInputStream s) throws java.io.IOException, ClassNotFoundException + { + // Read in capacity, and any hidden stuff + s.defaultReadObject(); + head = new Node<E>(null); + tail = head; + // Read in all elements and place in queue + for (;;) + { + @SuppressWarnings("unchecked") + E item = (E)s.readObject(); + if (item == null) + break; + else + offer(item); + } + } + + public static final AtomicReferenceFieldUpdater<HornetQConcurrentLinkedQueue, Node> tailFieldUpdater = AtomicReferenceFieldUpdater.newUpdater(HornetQConcurrentLinkedQueue.class, + Node.class, + "tail"); + + public static final AtomicReferenceFieldUpdater<HornetQConcurrentLinkedQueue, Node> headFieldUpdater = AtomicReferenceFieldUpdater.newUpdater(HornetQConcurrentLinkedQueue.class, + Node.class, + "head"); + + private boolean casTail(Node<E> cmp, Node<E> val) + { + return tailFieldUpdater.compareAndSet(this, cmp, val); + } + + private boolean casHead(Node<E> cmp, Node<E> val) + { + return headFieldUpdater.compareAndSet(this, cmp, val); + } + + private void lazySetHead(Node<E> val) + { + headFieldUpdater.lazySet(this, val); + } + +} Modified: trunk/tests/jms-tests/src/org/hornetq/jms/tests/message/JMSMessageIDHeaderTest.java =================================================================== --- trunk/tests/jms-tests/src/org/hornetq/jms/tests/message/JMSMessageIDHeaderTest.java 2010-03-22 23:43:26 UTC (rev 8946) +++ trunk/tests/jms-tests/src/org/hornetq/jms/tests/message/JMSMessageIDHeaderTest.java 2010-03-23 15:11:11 UTC (rev 8947) @@ -39,6 +39,7 @@ Message m = queueProducerSession.createMessage(); queueProducer.send(m); String messageID = queueConsumer.receive().getJMSMessageID(); + // JMS1.1 specs 3.4.3 ProxyAssertSupport.assertTrue(messageID.startsWith("ID:")); } Modified: trunk/tests/jms-tests/src/org/hornetq/jms/tests/message/MessageHeaderTest.java =================================================================== --- trunk/tests/jms-tests/src/org/hornetq/jms/tests/message/MessageHeaderTest.java 2010-03-22 23:43:26 UTC (rev 8946) +++ trunk/tests/jms-tests/src/org/hornetq/jms/tests/message/MessageHeaderTest.java 2010-03-23 15:11:11 UTC (rev 8947) @@ -310,7 +310,7 @@ while (en.hasMoreElements()) { String propName = (String)en.nextElement(); - + propNames.add(propName); } Modified: trunk/tests/jms-tests/src/org/hornetq/jms/tests/selector/SelectorTest.java =================================================================== --- trunk/tests/jms-tests/src/org/hornetq/jms/tests/selector/SelectorTest.java 2010-03-22 23:43:26 UTC (rev 8946) +++ trunk/tests/jms-tests/src/org/hornetq/jms/tests/selector/SelectorTest.java 2010-03-23 15:11:11 UTC (rev 8947) @@ -80,6 +80,8 @@ prod.send(redMessage); prod.send(blueMessage); + + log.info("sent message"); Message rec = redConsumer.receive(); ProxyAssertSupport.assertEquals(redMessage.getJMSMessageID(), rec.getJMSMessageID()); @@ -88,10 +90,12 @@ ProxyAssertSupport.assertNull(redConsumer.receive(3000)); redConsumer.close(); + + log.info("closed first consumer"); MessageConsumer universalConsumer = session.createConsumer(HornetQServerTestCase.queue1); - rec = universalConsumer.receive(); + rec = universalConsumer.receive(2000); ProxyAssertSupport.assertEquals(rec.getJMSMessageID(), blueMessage.getJMSMessageID()); ProxyAssertSupport.assertEquals("blue", rec.getStringProperty("color")); @@ -168,6 +172,7 @@ try { conn = getConnectionFactory().createConnection(); + conn.start(); Session sess = conn.createSession(false, Session.AUTO_ACKNOWLEDGE); @@ -181,12 +186,16 @@ Message m = sess.createMessage(); m.setStringProperty("beatle", "john"); + + m.setIntProperty("wibble", j); prod.send(m); m = sess.createMessage(); m.setStringProperty("beatle", "kermit the frog"); + + m.setIntProperty("wibble", j); prod.send(m); } @@ -194,10 +203,14 @@ for (int j = 0; j < 100; j++) { Message m = cons1.receive(1000); - + ProxyAssertSupport.assertNotNull(m); + + assertEquals(j, m.getIntProperty("wibble")); ProxyAssertSupport.assertEquals("john", m.getStringProperty("beatle")); + + log.info("Got message " + j); } Message m = cons1.receiveNoWait(); @@ -213,13 +226,20 @@ m = cons2.receive(1000); ProxyAssertSupport.assertNotNull(m); + + assertEquals(j, m.getIntProperty("wibble")); ProxyAssertSupport.assertEquals("kermit the frog", m.getStringProperty("beatle")); } - m = cons2.receiveNoWait(); +// m = cons2.receiveNoWait(); +// +// if (m != null) +// { +// log.info("got " + m.getStringProperty("beatle") + " j: " + m.getIntProperty("wibble")); +// } - ProxyAssertSupport.assertNull(m); + //ProxyAssertSupport.assertNull(m); } finally { @@ -528,7 +548,7 @@ } } } - + public void testManyConsumersWithDifferentSelectors() throws Exception { Connection conn = null; Modified: trunk/tests/src/org/hornetq/tests/integration/client/AutogroupIdTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/client/AutogroupIdTest.java 2010-03-22 23:43:26 UTC (rev 8946) +++ trunk/tests/src/org/hornetq/tests/integration/client/AutogroupIdTest.java 2010-03-23 15:11:11 UTC (rev 8947) @@ -24,6 +24,7 @@ import org.hornetq.api.core.client.ClientSession; import org.hornetq.api.core.client.ClientSessionFactory; import org.hornetq.api.core.client.MessageHandler; +import org.hornetq.core.logging.Logger; import org.hornetq.core.server.HornetQServer; import org.hornetq.tests.util.ServiceTestBase; @@ -32,6 +33,8 @@ */ public class AutogroupIdTest extends ServiceTestBase { + private static final Logger log = Logger.getLogger(AutogroupIdTest.class); + public final SimpleString addressA = new SimpleString("addressA"); public final SimpleString queueA = new SimpleString("queueA"); @@ -73,6 +76,8 @@ ClientConsumer consumer2 = session.createConsumer(groupTestQ); consumer2.setMessageHandler(myMessageHandler2); + log.info("starting session"); + session.start(); final int numMessages = 100; @@ -84,9 +89,11 @@ latch.await(); session.close(); + + log.info(myMessageHandler2.messagesReceived); - Assert.assertEquals(myMessageHandler.messagesReceived, 100); - Assert.assertEquals(myMessageHandler2.messagesReceived, 0); + Assert.assertEquals(100, myMessageHandler.messagesReceived); + Assert.assertEquals(0, myMessageHandler2.messagesReceived); } finally { @@ -200,8 +207,8 @@ session.close(); - Assert.assertEquals(myMessageHandler.messagesReceived, 50); - Assert.assertEquals(myMessageHandler2.messagesReceived, 50); + Assert.assertEquals(50, myMessageHandler.messagesReceived); + Assert.assertEquals(50, myMessageHandler2.messagesReceived); } finally { Modified: trunk/tests/src/org/hornetq/tests/integration/client/ConsumerTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/client/ConsumerTest.java 2010-03-22 23:43:26 UTC (rev 8946) +++ trunk/tests/src/org/hornetq/tests/integration/client/ConsumerTest.java 2010-03-23 15:11:11 UTC (rev 8947) @@ -65,6 +65,41 @@ super.tearDown(); } + +// public void testQueueSpin() throws Exception +// { +// ClientSessionFactory sf = createInVMFactory(); +// +// ClientSession session1 = sf.createSession(); +// +// ClientSession session2 = sf.createSession(); +// +// session1.createQueue(QUEUE, QUEUE, null, false); +// +// ClientProducer producer = session1.createProducer(QUEUE); +// +// final int numMessages = 100; +// +// for (int i = 0; i < numMessages; i++) +// { +// ClientMessage message = createTextMessage("m" + i, session1); +// producer.send(message); +// } +// +// ClientConsumer consumer1 = session1.createConsumer(QUEUE); +// +// ClientConsumer consumer2 = session2.createConsumer(QUEUE, new SimpleString("foo=wibble")); +// +// session2.start(); +// +// consumer2.receive(); +// +// Thread.sleep(30000); +// +// session1.close(); +// +// session2.close(); +// } public void testConsumerAckImmediateAutoCommitTrue() throws Exception { Deleted: trunk/tests/src/org/hornetq/tests/integration/client/CoreSelectorTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/client/CoreSelectorTest.java 2010-03-22 23:43:26 UTC (rev 8946) +++ trunk/tests/src/org/hornetq/tests/integration/client/CoreSelectorTest.java 2010-03-23 15:11:11 UTC (rev 8947) @@ -1,117 +0,0 @@ -/* - * Copyright 2010 Red Hat, Inc. - * Red Hat licenses this file to you under the Apache License, version - * 2.0 (the "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * http://www.apache.org/licenses/LICENSE-2.0 - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or - * implied. See the License for the specific language governing - * permissions and limitations under the License. - */ - -package org.hornetq.tests.integration.client; - -import org.hornetq.api.core.client.ClientConsumer; -import org.hornetq.api.core.client.ClientMessage; -import org.hornetq.api.core.client.ClientProducer; -import org.hornetq.api.core.client.ClientSession; -import org.hornetq.api.core.client.ClientSessionFactory; -import org.hornetq.core.server.HornetQServer; -import org.hornetq.tests.util.ServiceTestBase; - -/** - * A CoreSelectorTest - * - * @author <mailto:clebert.suconic@jboss.org">Clebert Suconic</a> - * - * - */ -public class CoreSelectorTest extends ServiceTestBase -{ - - // Constants ----------------------------------------------------- - - // Attributes ---------------------------------------------------- - - // Static -------------------------------------------------------- - - // Constructors -------------------------------------------------- - - // Public -------------------------------------------------------- - - public void testSelector() throws Exception - { - HornetQServer server = createServer(false, false); - - server.start(); - - ClientSessionFactory factory = createInVMFactory(); - ClientSession session = null; - - try - { - - session = factory.createSession(); - - session.createQueue("queue", "queue"); - ClientProducer prod = session.createProducer("queue"); - - - ClientMessage msg = session.createMessage(false); - msg.putIntProperty("intValue", 1); - - msg.putIntProperty("intValue", 1); - msg.putBytesProperty("bValue", new byte[]{'1'}); - - prod.send(msg); - - msg = session.createMessage(false); - msg.putIntProperty("intValue", 2); - - session.start(); - - ClientConsumer cons = session.createConsumer("queue", "bValue=1"); - - assertNull(cons.receiveImmediate()); - - cons.close(); - - cons = session.createConsumer("queue", "intValue=1"); - - msg = cons.receive(5000); - - assertNotNull(msg); - - assertEquals(1, (int)msg.getIntProperty("intValue")); - - assertNull(cons.receiveImmediate()); - - - - - - - } - finally - { - if (session != null) - { - session.close(); - } - - factory.close(); - server.stop(); - } - } - - // Package protected --------------------------------------------- - - // Protected ----------------------------------------------------- - - // Private ------------------------------------------------------- - - // Inner classes ------------------------------------------------- - -} Modified: trunk/tests/src/org/hornetq/tests/integration/client/ExpiryAddressTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/client/ExpiryAddressTest.java 2010-03-22 23:43:26 UTC (rev 8946) +++ trunk/tests/src/org/hornetq/tests/integration/client/ExpiryAddressTest.java 2010-03-23 15:11:11 UTC (rev 8947) @@ -58,7 +58,6 @@ ClientConsumer clientConsumer = clientSession.createConsumer(qName); ClientMessage m = clientConsumer.receiveImmediate(); Assert.assertNull(m); - System.out.println("size3 = " + server.getPostOffice().getPagingManager().getTotalMemory()); m = clientConsumer.receiveImmediate(); Assert.assertNull(m); clientConsumer.close(); @@ -67,9 +66,6 @@ Assert.assertNotNull(m); Assert.assertEquals(m.getBodyBuffer().readString(), "heyho!"); m.acknowledge(); - - // PageSize should be the same as when it started - Assert.assertEquals(0, server.getPostOffice().getPagingManager().getTotalMemory()); } public void testBasicSendToMultipleQueues() throws Exception @@ -88,20 +84,12 @@ ClientMessage clientMessage = createTextMessage("heyho!", clientSession); clientMessage.setExpiration(System.currentTimeMillis()); - System.out.println("initialPageSize = " + server.getPostOffice().getPagingManager().getTotalMemory()); - producer.send(clientMessage); - System.out.println("pageSize after message sent = " + server.getPostOffice().getPagingManager().getTotalMemory()); - clientSession.start(); ClientConsumer clientConsumer = clientSession.createConsumer(qName); ClientMessage m = clientConsumer.receiveImmediate(); - System.out.println("pageSize after message received = " + server.getPostOffice() - .getPagingManager() - .getTotalMemory()); - Assert.assertNull(m); clientConsumer.close(); @@ -133,9 +121,6 @@ clientConsumer.close(); clientSession.commit(); - - // PageGlobalSize should be untouched as the message expired - Assert.assertEquals(0, server.getPostOffice().getPagingManager().getTotalMemory()); } public void testBasicSendToNoQueue() throws Exception Modified: trunk/tests/src/org/hornetq/tests/integration/client/LargeMessageTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/client/LargeMessageTest.java 2010-03-22 23:43:26 UTC (rev 8946) +++ trunk/tests/src/org/hornetq/tests/integration/client/LargeMessageTest.java 2010-03-23 15:11:11 UTC (rev 8947) @@ -2133,7 +2133,6 @@ } } - assertGlobalSize(server); Assert.assertEquals(0, ((Queue)server.getPostOffice().getBinding(LargeMessageTest.ADDRESS).getBindable()).getDeliveringCount()); Assert.assertEquals(0, @@ -2238,7 +2237,6 @@ consumer.close(); } - assertGlobalSize(server); Assert.assertEquals(0, ((Queue)server.getPostOffice().getBinding(LargeMessageTest.ADDRESS).getBindable()).getDeliveringCount()); Assert.assertEquals(0, @@ -2311,7 +2309,6 @@ session.commit(); - assertGlobalSize(server); Assert.assertEquals(0, ((Queue)server.getPostOffice().getBinding(LargeMessageTest.ADDRESS).getBindable()).getDeliveringCount()); Assert.assertEquals(0, @@ -2397,7 +2394,6 @@ session.commit(); - assertGlobalSize(server); Assert.assertEquals(0, ((Queue)server.getPostOffice().getBinding(LargeMessageTest.ADDRESS).getBindable()).getDeliveringCount()); Assert.assertEquals(0, @@ -2627,18 +2623,6 @@ // Private ------------------------------------------------------- - private void assertGlobalSize(final HornetQServer server) throws InterruptedException - { - // addGlobalSize on LargeMessage is only done after the delivery, and the addSize could be asynchronous - long timeout = System.currentTimeMillis() + 5000; - while (timeout > System.currentTimeMillis() && server.getPostOffice().getPagingManager().getTotalMemory() != 0) - { - Thread.sleep(100); - } - - Assert.assertEquals(0l, server.getPostOffice().getPagingManager().getTotalMemory()); - } - // Inner classes ------------------------------------------------- } Modified: trunk/tests/src/org/hornetq/tests/integration/client/MessageGroupingConnectionFactoryTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/client/MessageGroupingConnectionFactoryTest.java 2010-03-22 23:43:26 UTC (rev 8946) +++ trunk/tests/src/org/hornetq/tests/integration/client/MessageGroupingConnectionFactoryTest.java 2010-03-23 15:11:11 UTC (rev 8947) @@ -44,43 +44,27 @@ public void testBasicGroupingUsingConnection() throws Exception { - doTestBasicGroupingUsingConnectionFactory(false); + doTestBasicGroupingUsingConnectionFactory(); } - public void testBasicGroupingUsingConnectionDirect() throws Exception - { - doTestBasicGroupingUsingConnectionFactory(true); - } - public void testBasicGroupingMultipleProducers() throws Exception { - doTestBasicGroupingMultipleProducers(false); + doTestBasicGroupingMultipleProducers(); } - public void testBasicGroupingMultipleProducersDirect() throws Exception + private void doTestBasicGroupingUsingConnectionFactory() throws Exception { - doTestBasicGroupingMultipleProducers(true); - } - - private void doTestBasicGroupingUsingConnectionFactory(final boolean directDelivery) throws Exception - { ClientProducer clientProducer = clientSession.createProducer(qName); ClientConsumer consumer = clientSession.createConsumer(qName); ClientConsumer consumer2 = clientSession.createConsumer(qName); - if (directDelivery) - { - clientSession.start(); - } + clientSession.start(); + int numMessages = 100; for (int i = 0; i < numMessages; i++) { ClientMessage message = createTextMessage("m" + i, clientSession); clientProducer.send(message); } - if (!directDelivery) - { - clientSession.start(); - } CountDownLatch latch = new CountDownLatch(numMessages); DummyMessageHandler dummyMessageHandler = new DummyMessageHandler(latch, true); consumer.setMessageHandler(dummyMessageHandler); @@ -93,17 +77,15 @@ consumer2.close(); } - private void doTestBasicGroupingMultipleProducers(final boolean directDelivery) throws Exception + private void doTestBasicGroupingMultipleProducers() throws Exception { ClientProducer clientProducer = clientSession.createProducer(qName); ClientProducer clientProducer2 = clientSession.createProducer(qName); ClientProducer clientProducer3 = clientSession.createProducer(qName); ClientConsumer consumer = clientSession.createConsumer(qName); ClientConsumer consumer2 = clientSession.createConsumer(qName); - if (directDelivery) - { - clientSession.start(); - } + clientSession.start(); + int numMessages = 100; for (int i = 0; i < numMessages; i++) { @@ -112,10 +94,6 @@ clientProducer2.send(message); clientProducer3.send(message); } - if (!directDelivery) - { - clientSession.start(); - } CountDownLatch latch = new CountDownLatch(numMessages * 3); DummyMessageHandler dummyMessageHandler = new DummyMessageHandler(latch, true); consumer.setMessageHandler(dummyMessageHandler); Modified: trunk/tests/src/org/hornetq/tests/integration/client/MessageGroupingTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/client/MessageGroupingTest.java 2010-03-22 23:43:26 UTC (rev 8946) +++ trunk/tests/src/org/hornetq/tests/integration/client/MessageGroupingTest.java 2010-03-23 15:11:11 UTC (rev 8947) @@ -46,26 +46,16 @@ private final SimpleString qName = new SimpleString("MessageGroupingTestQueue"); - public void testBasicGroupingWithDirectDelivery() throws Exception + public void testBasicGrouping() throws Exception { - doTestBasicGrouping(true); + doTestBasicGrouping(); } - public void testBasicGroupingWithoutDirectDelivery() throws Exception + public void testMultipleGrouping() throws Exception { - doTestBasicGrouping(false); + doTestMultipleGrouping(); } - public void testMultipleGroupingWithDirectDelivery() throws Exception - { - doTestMultipleGrouping(true); - } - - public void testMultipleGroupingWithoutDirectDelivery() throws Exception - { - doTestMultipleGrouping(false); - } - public void testMultipleGroupingSingleConsumerWithDirectDelivery() throws Exception { doTestMultipleGroupingSingleConsumer(true); @@ -76,55 +66,33 @@ doTestMultipleGroupingSingleConsumer(false); } - public void testMultipleGroupingTXCommitWithDirectDelivery() throws Exception + public void testMultipleGroupingTXCommit() throws Exception { - doTestMultipleGroupingTXCommit(true); + doTestMultipleGroupingTXCommit(); } - public void testMultipleGroupingTXCommitWithoutDirectDelivery() throws Exception + public void testMultipleGroupingTXRollback() throws Exception { - doTestMultipleGroupingTXCommit(false); + doTestMultipleGroupingTXRollback(); } - public void testMultipleGroupingTXRollbackWithDirectDelivery() throws Exception + public void testMultipleGroupingXACommit() throws Exception { - doTestMultipleGroupingTXRollback(true); + dotestMultipleGroupingXACommit(); } - public void testMultipleGroupingTXRollbackWithoutDirectDelivery() throws Exception + public void testMultipleGroupingXARollback() throws Exception { - doTestMultipleGroupingTXRollback(false); + doTestMultipleGroupingXARollback(); } - public void testMultipleGroupingXACommitWithDirectDelivery() throws Exception + private void doTestBasicGrouping() throws Exception { - dotestMultipleGroupingXACommit(true); - } - - public void testMultipleGroupingXACommitWithoutDirectDelivery() throws Exception - { - dotestMultipleGroupingXACommit(false); - } - - public void testMultipleGroupingXARollbackWithDirectDelivery() throws Exception - { - doTestMultipleGroupingXARollback(true); - } - - public void testMultipleGroupingXARollbackWithoutDirectDelivery() throws Exception - { - doTestMultipleGroupingXARollback(false); - } - - private void doTestBasicGrouping(final boolean directDelivery) throws Exception - { ClientProducer clientProducer = clientSession.createProducer(qName); ClientConsumer consumer = clientSession.createConsumer(qName); ClientConsumer consumer2 = clientSession.createConsumer(qName); - if (directDelivery) - { - clientSession.start(); - } + clientSession.start(); + SimpleString groupId = new SimpleString("grp1"); int numMessages = 100; for (int i = 0; i < numMessages; i++) @@ -133,10 +101,7 @@ message.putStringProperty(Message.HDR_GROUP_ID, groupId); clientProducer.send(message); } - if (!directDelivery) - { - clientSession.start(); - } + CountDownLatch latch = new CountDownLatch(numMessages); DummyMessageHandler dummyMessageHandler = new DummyMessageHandler(latch, true); consumer.setMessageHandler(dummyMessageHandler); @@ -155,6 +120,10 @@ ClientConsumer consumer = clientSession.createConsumer(qName); ClientConsumer consumer2 = clientSession.createConsumer(qName); clientSession.start(); + + //need to wait a bit or consumers might be busy + Thread.sleep(200); + SimpleString groupId = new SimpleString("grp1"); SimpleString groupId2 = new SimpleString("grp2"); int numMessages = 100; @@ -183,30 +152,11 @@ Assert.assertEquals(cm.getBodyBuffer().readString(), "m" + i); } - MessageGroupingTest.log.info("closing consumers"); + MessageGroupingTest.log.info("closing consumer2"); consumer2.close(); - MessageGroupingTest.log.info("closed consumer 2"); - consumer.close(); - - MessageGroupingTest.log.info("closed consuemrs"); - // check that within their groups the messages are still in the correct order - consumer = clientSession.createConsumer(qName); - for (int i = 0; i < numMessages; i += 2) - { - ClientMessage cm = consumer.receive(500); - Assert.assertNotNull(cm); - Assert.assertEquals(cm.getBodyBuffer().readString(), "m" + i); - } - for (int i = 1; i < numMessages; i += 2) - { - ClientMessage cm = consumer.receive(500); - Assert.assertNotNull(cm); - Assert.assertEquals(cm.getBodyBuffer().readString(), "m" + i); - } - consumer.close(); } private void doTestMultipleGroupingSingleConsumer(final boolean directDelivery) throws Exception @@ -251,17 +201,18 @@ consumer.close(); } - private void doTestMultipleGroupingTXCommit(final boolean directDelivery) throws Exception + private void doTestMultipleGroupingTXCommit() throws Exception { ClientSessionFactory sessionFactory = HornetQClient.createClientSessionFactory(new TransportConfiguration(UnitTestCase.INVM_CONNECTOR_FACTORY)); ClientSession clientSession = sessionFactory.createSession(false, false, false); ClientProducer clientProducer = this.clientSession.createProducer(qName); - if (directDelivery) - { - clientSession.start(); - } + clientSession.start(); + ClientConsumer consumer = clientSession.createConsumer(qName); ClientConsumer consumer2 = clientSession.createConsumer(qName); + + //Wait a bit otherwise consumers might be busy + Thread.sleep(200); SimpleString groupId = new SimpleString("grp1"); SimpleString groupId2 = new SimpleString("grp2"); @@ -279,10 +230,7 @@ } clientProducer.send(message); } - if (!directDelivery) - { - clientSession.start(); - } + CountDownLatch latch = new CountDownLatch(numMessages); DummyMessageHandler dummyMessageHandler = new DummyMessageHandler(latch, true); consumer.setMessageHandler(dummyMessageHandler); @@ -311,18 +259,20 @@ clientSession.close(); } - private void doTestMultipleGroupingTXRollback(final boolean directDelivery) throws Exception + private void doTestMultipleGroupingTXRollback() throws Exception { + log.info("*** starting test"); ClientSessionFactory sessionFactory = HornetQClient.createClientSessionFactory(new TransportConfiguration(UnitTestCase.INVM_CONNECTOR_FACTORY)); sessionFactory.setBlockOnAcknowledge(true); ClientSession clientSession = sessionFactory.createSession(false, false, false); ClientProducer clientProducer = this.clientSession.createProducer(qName); ClientConsumer consumer = clientSession.createConsumer(qName); ClientConsumer consumer2 = clientSession.createConsumer(qName); - if (directDelivery) - { - clientSession.start(); - } + clientSession.start(); + + //need to wait a bit or consumers might be busy + Thread.sleep(200); + SimpleString groupId = new SimpleString("grp1"); SimpleString groupId2 = new SimpleString("grp2"); int numMessages = 100; @@ -339,17 +289,14 @@ } clientProducer.send(message); } - if (!directDelivery) - { - clientSession.start(); - } + CountDownLatch latch = new CountDownLatch(numMessages); DummyMessageHandler dummyMessageHandler = new DummyMessageHandler(latch, true); consumer.setMessageHandler(dummyMessageHandler); DummyMessageHandler dummyMessageHandler2 = new DummyMessageHandler(latch, true); consumer2.setMessageHandler(dummyMessageHandler2); Assert.assertTrue(latch.await(10, TimeUnit.SECONDS)); - Assert.assertEquals(dummyMessageHandler.list.size(), 50); + Assert.assertEquals(50, dummyMessageHandler.list.size(), dummyMessageHandler.list.size()); int i = 0; for (ClientMessage message : dummyMessageHandler.list) { @@ -387,17 +334,15 @@ clientSession.close(); } - private void dotestMultipleGroupingXACommit(final boolean directDelivery) throws Exception + private void dotestMultipleGroupingXACommit() throws Exception { ClientSessionFactory sessionFactory = HornetQClient.createClientSessionFactory(new TransportConfiguration(UnitTestCase.INVM_CONNECTOR_FACTORY)); ClientSession clientSession = sessionFactory.createSession(true, false, false); ClientProducer clientProducer = this.clientSession.createProducer(qName); ClientConsumer consumer = clientSession.createConsumer(qName); ClientConsumer consumer2 = clientSession.createConsumer(qName); - if (directDelivery) - { - clientSession.start(); - } + clientSession.start(); + Xid xid = new XidImpl("bq".getBytes(), 4, "gtid".getBytes()); clientSession.start(xid, XAResource.TMNOFLAGS); @@ -417,10 +362,6 @@ } clientProducer.send(message); } - if (!directDelivery) - { - clientSession.start(); - } CountDownLatch latch = new CountDownLatch(numMessages); DummyMessageHandler dummyMessageHandler = new DummyMessageHandler(latch, true); consumer.setMessageHandler(dummyMessageHandler); @@ -451,16 +392,14 @@ clientSession.close(); } - private void doTestMultipleGroupingXARollback(final boolean directDelivery) throws Exception + private void doTestMultipleGroupingXARollback() throws Exception { ClientSessionFactory sessionFactory = HornetQClient.createClientSessionFactory(new TransportConfiguration(UnitTestCase.INVM_CONNECTOR_FACTORY)); sessionFactory.setBlockOnAcknowledge(true); ClientSession clientSession = sessionFactory.createSession(true, false, false); ClientProducer clientProducer = this.clientSession.createProducer(qName); - if (directDelivery) - { - clientSession.start(); - } + clientSession.start(); + ClientConsumer consumer = clientSession.createConsumer(qName); ClientConsumer consumer2 = clientSession.createConsumer(qName); Xid xid = new XidImpl("bq".getBytes(), 4, "gtid".getBytes()); @@ -482,10 +421,7 @@ } clientProducer.send(message); } - if (!directDelivery) - { - clientSession.start(); - } + CountDownLatch latch = new CountDownLatch(numMessages); DummyMessageHandler dummyMessageHandler = new DummyMessageHandler(latch, true); consumer.setMessageHandler(dummyMessageHandler); @@ -535,18 +471,16 @@ clientSession.close(); } - private void doTestMultipleGrouping(final boolean directDelivery) throws Exception + private void doTestMultipleGrouping() throws Exception { ClientProducer clientProducer = clientSession.createProducer(qName); ClientConsumer consumer = clientSession.createConsumer(qName); ClientConsumer consumer2 = clientSession.createConsumer(qName); - if (directDelivery) - { - clientSession.start(); - } + clientSession.start(); + SimpleString groupId = new SimpleString("grp1"); SimpleString groupId2 = new SimpleString("grp2"); - int numMessages = 100; + int numMessages = 10; for (int i = 0; i < numMessages; i++) { ClientMessage message = createTextMessage("m" + i, clientSession); @@ -560,24 +494,21 @@ } clientProducer.send(message); } - if (!directDelivery) - { - clientSession.start(); - } + CountDownLatch latch = new CountDownLatch(numMessages); DummyMessageHandler dummyMessageHandler = new DummyMessageHandler(latch, true); consumer.setMessageHandler(dummyMessageHandler); DummyMessageHandler dummyMessageHandler2 = new DummyMessageHandler(latch, true); consumer2.setMessageHandler(dummyMessageHandler2); Assert.assertTrue(latch.await(10, TimeUnit.SECONDS)); - Assert.assertEquals(50, dummyMessageHandler.list.size()); + Assert.assertEquals(numMessages / 2, dummyMessageHandler.list.size()); int i = 0; for (ClientMessage message : dummyMessageHandler.list) { Assert.assertEquals(message.getBodyBuffer().readString(), "m" + i); i += 2; } - Assert.assertEquals(50, dummyMessageHandler2.list.size()); + Assert.assertEquals(numMessages / 2, dummyMessageHandler2.list.size()); i = 1; for (ClientMessage message : dummyMessageHandler2.list) { Modified: trunk/tests/src/org/hornetq/tests/integration/client/MessageHandlerTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/client/MessageHandlerTest.java 2010-03-22 23:43:26 UTC (rev 8946) +++ trunk/tests/src/org/hornetq/tests/integration/client/MessageHandlerTest.java 2010-03-23 15:11:11 UTC (rev 8947) @@ -17,6 +17,7 @@ import junit.framework.Assert; +import org.hornetq.api.core.HornetQBuffer; import org.hornetq.api.core.SimpleString; import org.hornetq.api.core.client.ClientConsumer; import org.hornetq.api.core.client.ClientMessage; @@ -135,14 +136,14 @@ for (int i = 0; i < numMessages; i++) { ClientMessage message = createTextMessage("m" + i, session); + message.putIntProperty(new SimpleString("i"), i); + producer.send(message); } final ClientConsumer consumer = session.createConsumer(QUEUE); - session.start(); - CountDownLatch latch = new CountDownLatch(50); // Message should be in consumer @@ -171,14 +172,19 @@ { failed = true; } + messageReceived++; + + log.info("got message " + messageReceived); + latch.countDown(); if (latch.getCount() == 0) { - message.acknowledge(); + started = false; + consumer.setMessageHandler(null); } @@ -192,7 +198,10 @@ MyHandler handler = new MyHandler(latch); consumer.setMessageHandler(handler); + + session.start(); + latch.await(); Thread.sleep(100); Modified: trunk/tests/src/org/hornetq/tests/integration/client/MessagePriorityTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/client/MessagePriorityTest.java 2010-03-22 23:43:26 UTC (rev 8946) +++ trunk/tests/src/org/hornetq/tests/integration/client/MessagePriorityTest.java 2010-03-23 15:11:11 UTC (rev 8947) @@ -17,9 +17,15 @@ import org.hornetq.api.core.SimpleString; import org.hornetq.api.core.TransportConfiguration; -import org.hornetq.api.core.client.*; +import org.hornetq.api.core.client.ClientConsumer; +import org.hornetq.api.core.client.ClientMessage; +import org.hornetq.api.core.client.ClientProducer; +import org.hornetq.api.core.client.ClientSession; +import org.hornetq.api.core.client.ClientSessionFactory; +import org.hornetq.api.core.client.HornetQClient; import org.hornetq.core.config.Configuration; import org.hornetq.core.config.impl.ConfigurationImpl; +import org.hornetq.core.logging.Logger; import org.hornetq.core.remoting.impl.invm.InVMAcceptorFactory; import org.hornetq.core.remoting.impl.invm.InVMConnectorFactory; import org.hornetq.core.server.HornetQServer; @@ -37,6 +43,9 @@ // Constants ----------------------------------------------------- + private static final Logger log = Logger.getLogger(MessagePriorityTest.class); + + // Attributes ---------------------------------------------------- private HornetQServer server; @@ -95,9 +104,11 @@ SimpleString address = RandomUtil.randomSimpleString(); session.createQueue(address, queue, false); + + session.start(); ClientProducer producer = session.createProducer(address); - session.start(); + ClientConsumer consumer = session.createConsumer(queue); for (int i = 0; i < 10; i++) @@ -106,8 +117,12 @@ m.setPriority((byte)i); producer.send(m); } - - // expect to consumer message with higher priority first + + //Wait for msgs to reach client side + + Thread.sleep(1000); + + // expect to consume message with higher priority first for (int i = 9; i >= 0; i--) { ClientMessage m = consumer.receive(500); Added: trunk/tests/src/org/hornetq/tests/integration/client/NIOvsOIOTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/client/NIOvsOIOTest.java (rev 0) +++ trunk/tests/src/org/hornetq/tests/integration/client/NIOvsOIOTest.java 2010-03-23 15:11:11 UTC (rev 8947) @@ -0,0 +1,339 @@ +/* + * Copyright 2009 Red Hat, Inc. + * Red Hat licenses this file to you under the Apache License, version + * 2.0 (the "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * http://www.apache.org/licenses/LICENSE-2.0 + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or + * implied. See the License for the specific language governing + * permissions and limitations under the License. + */ + +package org.hornetq.tests.integration.client; + +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.CountDownLatch; + +import junit.framework.TestSuite; + +import org.hornetq.api.core.TransportConfiguration; +import org.hornetq.api.core.client.ClientConsumer; +import org.hornetq.api.core.client.ClientMessage; +import org.hornetq.api.core.client.ClientProducer; +import org.hornetq.api.core.client.ClientSession; +import org.hornetq.api.core.client.ClientSessionFactory; +import org.hornetq.api.core.client.HornetQClient; +import org.hornetq.api.core.client.MessageHandler; +import org.hornetq.core.config.Configuration; +import org.hornetq.core.config.impl.ConfigurationImpl; +import org.hornetq.core.logging.Logger; +import org.hornetq.core.server.HornetQServer; +import org.hornetq.core.server.HornetQServers; +import org.hornetq.core.settings.HierarchicalRepository; +import org.hornetq.core.settings.impl.AddressFullMessagePolicy; +import org.hornetq.core.settings.impl.AddressSettings; +import org.hornetq.integration.transports.netty.TransportConstants; +import org.hornetq.tests.util.UnitTestCase; +import org.hornetq.utils.UUIDGenerator; + +public class NIOvsOIOTest extends UnitTestCase +{ + private static final Logger log = Logger.getLogger(NIOvsOIOTest.class); + + // Constants ----------------------------------------------------- + + // Attributes ---------------------------------------------------- + + // Static -------------------------------------------------------- + + // Constructors -------------------------------------------------- + + // Public -------------------------------------------------------- + + public static TestSuite suite() + { + return new TestSuite(); + } + +// public void testNIOPerf() throws Exception +// { +// log.info("************* Testing NIO"); +// testPerf(true); +// } +// +// public void testOIOPerf() throws Exception +// { +// log.info("************ Testing OIO"); +// testPerf(false); +// } + + private void doTest(String dest) throws Exception + { + System.getProperties().put("hq.batchHQ", "true"); + + String connectorFactoryClassName = "org.hornetq.integration.transports.netty.NettyConnectorFactory"; + + + final int numSenders = 1; + + final int numReceivers = 1; + + final int numMessages = 200000; + + Receiver[] receivers = new Receiver[numReceivers]; + + Sender[] senders = new Sender[numSenders]; + + List<ClientSessionFactory> factories = new ArrayList<ClientSessionFactory>(); + + for (int i = 0; i < numReceivers; i++) + { + ClientSessionFactory sf = HornetQClient.createClientSessionFactory(new TransportConfiguration(connectorFactoryClassName)); + + factories.add(sf); + + receivers[i] = new Receiver(i, sf, numMessages * numSenders, dest); + + receivers[i].prepare(); + + receivers[i].start(); + } + + for (int i = 0; i < numSenders; i++) + { + ClientSessionFactory sf = HornetQClient.createClientSessionFactory(new TransportConfiguration(connectorFactoryClassName)); + + factories.add(sf); + + senders[i] = new Sender(i, sf, numMessages, dest); + + senders[i].prepare(); + } + + long start = System.currentTimeMillis(); + + for (int i = 0; i < numSenders; i++) + { + senders[i].start(); + } + + for (int i = 0; i < numSenders; i++) + { + senders[i].join(); + } + + for (int i = 0; i < numReceivers; i++) + { + receivers[i].await(); + } + + long end = System.currentTimeMillis(); + + double rate = 1000 * (double)(numMessages * numSenders) / (end - start); + + log.info("Rate is " + rate + " msgs sec"); + + for (int i = 0; i < numSenders; i++) + { + senders[i].terminate(); + } + + for (int i = 0; i < numReceivers; i++) + { + receivers[i].terminate(); + } + + for (ClientSessionFactory sf: factories) + { + sf.close(); + } + } + + private void testPerf(boolean nio) throws Exception + { + String acceptorFactoryClassName = "org.hornetq.integration.transports.netty.NettyAcceptorFactory"; + + Configuration conf = new ConfigurationImpl(); + + conf.setSecurityEnabled(false); + + Map<String, Object> params = new HashMap<String, Object>(); + + params.put(TransportConstants.USE_NIO_PROP_NAME, nio); + + conf.getAcceptorConfigurations().add(new TransportConfiguration(acceptorFactoryClassName, params)); + + HornetQServer server = HornetQServers.newHornetQServer(conf, false); + + AddressSettings addressSettings = new AddressSettings(); + + addressSettings.setAddressFullMessagePolicy(AddressFullMessagePolicy.BLOCK); + + addressSettings.setMaxSizeBytes(10 * 1024 * 1024); + + final String dest = "test-destination"; + + HierarchicalRepository<AddressSettings> repos = server.getAddressSettingsRepository(); + + repos.addMatch(dest, addressSettings); + + server.start(); + + for (int i = 0; i < 2; i++) + { + doTest(dest); + } + + server.stop(); + } + + private class Sender extends Thread + { + private ClientSessionFactory sf; + + private int numMessages; + + private String dest; + + private ClientSession session; + + private ClientProducer producer; + + private int id; + + Sender(int id, ClientSessionFactory sf, final int numMessages, final String dest) + { + this.id = id; + + this.sf = sf; + + this.numMessages = numMessages; + + this.dest = dest; + } + + void prepare() throws Exception + { + session = sf.createSession(); + + producer = session.createProducer(dest); + } + + public void run() + { + ClientMessage msg = session.createMessage(false); + + for (int i = 0; i < numMessages; i++) + { + try + { + producer.send(msg); + } + catch (Exception e) + { + log.error("Caught exception", e); + } + + //log.info(id + " sent message " + i); + + } + } + + public void terminate() throws Exception + { + session.close(); + } + } + + private class Receiver implements MessageHandler + { + private ClientSessionFactory sf; + + private int numMessages; + + private String dest; + + private ClientSession session; + + private ClientConsumer consumer; + + private int id; + + private String queueName; + + Receiver(int id, ClientSessionFactory sf, final int numMessages, final String dest) + { + this.id = id; + + this.sf = sf; + + this.numMessages = numMessages; + + this.dest = dest; + } + + void prepare() throws Exception + { + session = sf.createSession(); + + queueName = UUIDGenerator.getInstance().generateStringUUID(); + + session.createQueue(dest, queueName); + + consumer = session.createConsumer(queueName); + + consumer.setMessageHandler(this); + } + + void start() throws Exception + { + session.start(); + } + + private CountDownLatch latch = new CountDownLatch(1); + + void await() throws Exception + { + latch.await(); + } + + private int count; + + public void onMessage(ClientMessage msg) + { + try + { + msg.acknowledge(); + } + catch (Exception e) + { + log.error("Caught exception", e); + } + + count++; + + if (count == numMessages) + { + latch.countDown(); + } + + //log.info(id + " got msg " + count); + + } + + public void terminate() throws Exception + { + consumer.close(); + + session.deleteQueue(queueName); + + session.close(); + } + } + +} Modified: trunk/tests/src/org/hornetq/tests/integration/client/OrderTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/client/OrderTest.java 2010-03-22 23:43:26 UTC (rev 8946) +++ trunk/tests/src/org/hornetq/tests/integration/client/OrderTest.java 2010-03-23 15:11:11 UTC (rev 8947) @@ -20,6 +20,7 @@ import org.hornetq.api.core.client.ClientProducer; import org.hornetq.api.core.client.ClientSession; import org.hornetq.api.core.client.ClientSessionFactory; +import org.hornetq.core.logging.Logger; import org.hornetq.core.server.HornetQServer; import org.hornetq.tests.util.ServiceTestBase; @@ -34,7 +35,10 @@ { // Constants ----------------------------------------------------- + + private static final Logger log = Logger.getLogger(OrderTest.class); + // Attributes ---------------------------------------------------- private HornetQServer server; @@ -119,6 +123,7 @@ if (!started || started && i % 2 == 0) { ClientMessage msg = cons.receive(10000); + Assert.assertEquals(i, msg.getIntProperty("id").intValue()); } } @@ -132,6 +137,7 @@ if (!started || started && i % 2 == 0) { ClientMessage msg = cons.receive(10000); + Assert.assertEquals(i, msg.getIntProperty("id").intValue()); } } Modified: trunk/tests/src/org/hornetq/tests/integration/client/PagingTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/client/PagingTest.java 2010-03-22 23:43:26 UTC (rev 8946) +++ trunk/tests/src/org/hornetq/tests/integration/client/PagingTest.java 2010-03-23 15:11:11 UTC (rev 8947) @@ -139,10 +139,6 @@ session.close(); - Assert.assertTrue("TotalMemory expected to be > 0 when it was " + server.getPostOffice() - .getPagingManager() - .getTotalMemory(), - server.getPostOffice().getPagingManager().getTotalMemory() > 0); server.stop(); @@ -155,11 +151,6 @@ sf = createInVMFactory(); - Assert.assertTrue("TotalMemory expected to be > 0 when it was " + server.getPostOffice() - .getPagingManager() - .getTotalMemory(), - server.getPostOffice().getPagingManager().getTotalMemory() > 0); - session = sf.createSession(null, null, false, true, true, false, 0); ClientConsumer consumer = session.createConsumer(PagingTest.ADDRESS); @@ -198,9 +189,6 @@ consumer.close(); session.close(); - - Assert.assertEquals(0, server.getPostOffice().getPagingManager().getTotalMemory()); - } finally { @@ -346,9 +334,6 @@ consumer.close(); session.close(); - - Assert.assertEquals(0, server.getPostOffice().getPagingManager().getTotalMemory()); - } finally { @@ -492,9 +477,6 @@ consumer.close(); session.close(); - - Assert.assertEquals(0, server.getPostOffice().getPagingManager().getTotalMemory()); - } finally { @@ -541,8 +523,6 @@ ClientProducer producer = session.createProducer(PagingTest.ADDRESS); - long initialSize = server.getPostOffice().getPagingManager().getTotalMemory(); - ClientMessage message = null; for (int i = 0; i < numberOfMessages; i++) @@ -570,8 +550,6 @@ Assert.assertNull(consumer.receiveImmediate()); session.close(); - - Assert.assertEquals(initialSize, server.getPostOffice().getPagingManager().getTotalMemory()); } finally { @@ -618,8 +596,6 @@ ClientProducer producer = session.createProducer(PagingTest.ADDRESS); - long initialSize = server.getPostOffice().getPagingManager().getTotalMemory(); - ClientMessage message = null; for (int i = 0; i < numberOfMessages; i++) @@ -653,8 +629,6 @@ session.commit(); session.close(); - - Assert.assertEquals(initialSize, server.getPostOffice().getPagingManager().getTotalMemory()); } finally { @@ -739,7 +713,6 @@ Assert.assertNull(consumer.receiveImmediate()); - Assert.assertEquals(0, server.getPostOffice().getPagingManager().getTotalMemory()); Assert.assertEquals(0, server.getPostOffice() .getPagingManager() .getPageStore(PagingTest.ADDRESS) @@ -803,7 +776,6 @@ session.close(); - Assert.assertEquals(0, server.getPostOffice().getPagingManager().getTotalMemory()); Assert.assertEquals(0, server.getPostOffice() .getPagingManager() .getPageStore(PagingTest.ADDRESS) @@ -985,8 +957,6 @@ sf = createInVMFactory(); - Assert.assertTrue(server.getPostOffice().getPagingManager().getTotalMemory() > 0); - session = sf.createSession(null, null, false, true, true, false, 0); session.start(); @@ -1021,10 +991,6 @@ Assert.assertEquals("Queue someQueue" + i + " was supposed to be empty", 0, queue.getDeliveringCount()); } - Assert.assertEquals("There are pending messages on the server", 0, server.getPostOffice() - .getPagingManager() - .getTotalMemory()); - } finally { Modified: trunk/tests/src/org/hornetq/tests/integration/client/ProducerFlowControlTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/client/ProducerFlowControlTest.java 2010-03-22 23:43:26 UTC (rev 8946) +++ trunk/tests/src/org/hornetq/tests/integration/client/ProducerFlowControlTest.java 2010-03-23 15:11:11 UTC (rev 8947) @@ -32,9 +32,7 @@ import org.hornetq.core.client.impl.ClientProducerInternal; import org.hornetq.core.client.impl.ClientSessionInternal; import org.hornetq.core.logging.Logger; -import org.hornetq.core.paging.impl.TestSupportPageStore; import org.hornetq.core.server.HornetQServer; -import org.hornetq.core.server.impl.ServerProducerCreditManager; import org.hornetq.core.settings.HierarchicalRepository; import org.hornetq.core.settings.impl.AddressFullMessagePolicy; import org.hornetq.core.settings.impl.AddressSettings; @@ -175,6 +173,8 @@ false); } + // (1000, 10 * 1024, 10 * 1024, 1024, 1024, 1024, 1, 1, 0, false); + private void testFlowControl(final int numMessages, final int messageSize, final int maxSize, @@ -309,15 +309,11 @@ else { producers[j].send(message); - - // log.info("sent message " + i); } } } - // log.info("sent messages"); - for (int i = 0; i < numConsumers; i++) { handlers[i].latch.await(); @@ -333,299 +329,9 @@ session.close(); - TestSupportPageStore store = (TestSupportPageStore)server.getPostOffice() - .getPagingManager() - .getPageStore(address); - - Assert.assertFalse(store.isExceededAvailableCredits()); - server.stop(); } - public void testUnusedCreditsAreReturnedOnSessionCloseWithWaitingEntries() throws Exception - { - final SimpleString address = new SimpleString("testaddress"); - - HornetQServer server = createServer(false, isNetty()); - - AddressSettings addressSettings = new AddressSettings(); - addressSettings.setMaxSizeBytes(4000); - addressSettings.setAddressFullMessagePolicy(AddressFullMessagePolicy.BLOCK); - - HierarchicalRepository<AddressSettings> repos = server.getAddressSettingsRepository(); - repos.addMatch(address.toString(), addressSettings); - - server.start(); - - ClientSessionFactory sf = createFactory(isNetty()); - - // Make sure the producer grabs all the credits - sf.setProducerWindowSize(4000); - sf.setConsumerWindowSize(1024); - sf.setAckBatchSize(1024); - - final ClientSession session = sf.createSession(false, true, true, true); - - final SimpleString queueName = new SimpleString("testqueue"); - - session.createQueue(address, queueName, null, false); - - ClientProducer producer = session.createProducer(address); - - ClientSession session2 = sf.createSession(false, true, true, true); - - ClientProducer producer2 = session2.createProducer(address); - - ServerProducerCreditManager mgr = server.getPostOffice() - .getPagingManager() - .getPageStore(address) - .getProducerCreditManager(); - - long start = System.currentTimeMillis(); - - int waiting; - do - { - waiting = mgr.waitingEntries(); - - Thread.sleep(10); - } - while (waiting != 1 || System.currentTimeMillis() - start > 3000); - - Assert.assertEquals(1, waiting); - - byte[] bytes = new byte[0]; - - ClientMessage message = session.createMessage(false); - - message.getBodyBuffer().writeBytes(bytes); - - producer.send(message); - - class SessionCloser implements Runnable - { - public void run() - { - try - { - Thread.sleep(2000); - - closed = true; - - session.close(); - } - catch (Exception e) - { - } - } - - volatile boolean closed; - } - - SessionCloser closer = new SessionCloser(); - - Thread t = new Thread(closer); - - t.start(); - - ClientMessage message2 = session.createMessage(false); - - message2.getBodyBuffer().writeBytes(bytes); - - producer2.send(message2); - - // Make sure it blocked until the first producer was closed - Assert.assertTrue(closer.closed); - - t.join(); - - session2.close(); - - TestSupportPageStore store = (TestSupportPageStore)server.getPostOffice() - .getPagingManager() - .getPageStore(address); - - Assert.assertFalse(store.isExceededAvailableCredits()); - - server.stop(); - } - - public void testUnusedCreditsAreReturnedOnSessionCloseNoWaitingEntries() throws Exception - { - final SimpleString address = new SimpleString("testaddress"); - - HornetQServer server = createServer(false, isNetty()); - - AddressSettings addressSettings = new AddressSettings(); - addressSettings.setMaxSizeBytes(4000); - addressSettings.setAddressFullMessagePolicy(AddressFullMessagePolicy.BLOCK); - - HierarchicalRepository<AddressSettings> repos = server.getAddressSettingsRepository(); - repos.addMatch(address.toString(), addressSettings); - - server.start(); - - ClientSessionFactory sf = createFactory(isNetty()); - - // Make sure producer grabs all the credits - sf.setProducerWindowSize(4000); - sf.setConsumerWindowSize(1024); - sf.setAckBatchSize(1024); - - final ClientSession session = sf.createSession(false, true, true, true); - - final SimpleString queueName = new SimpleString("testqueue"); - - session.createQueue(address, queueName, null, false); - - ClientProducer producer = session.createProducer(address); - - byte[] bytes = new byte[0]; - - ClientMessage message = session.createMessage(false); - - message.getBodyBuffer().writeBytes(bytes); - - producer.send(message); - - session.close(); - - ClientSession session2 = sf.createSession(false, true, true, true); - - ClientProducer producer2 = session2.createProducer(address); - - ServerProducerCreditManager mgr = server.getPostOffice() - .getPagingManager() - .getPageStore(address) - .getProducerCreditManager(); - - long start = System.currentTimeMillis(); - - int waiting; - do - { - waiting = mgr.waitingEntries(); - - Thread.sleep(10); - } - while (waiting != 1 || System.currentTimeMillis() - start > 3000); - - Assert.assertEquals(1, waiting); - - message = session.createMessage(false); - - message.getBodyBuffer().writeBytes(bytes); - - producer2.send(message); - - session2.close(); - - TestSupportPageStore store = (TestSupportPageStore)server.getPostOffice() - .getPagingManager() - .getPageStore(address); - - Assert.assertFalse(store.isExceededAvailableCredits()); - - server.stop(); - } - - public void testUnusedCreditsAreReturnedWaitingEntriesForClosedSessions() throws Exception - { - final SimpleString address = new SimpleString("testaddress"); - - HornetQServer server = createServer(false, isNetty()); - - AddressSettings addressSettings = new AddressSettings(); - addressSettings.setMaxSizeBytes(4000); - addressSettings.setAddressFullMessagePolicy(AddressFullMessagePolicy.BLOCK); - - HierarchicalRepository<AddressSettings> repos = server.getAddressSettingsRepository(); - repos.addMatch(address.toString(), addressSettings); - - server.start(); - - ClientSessionFactory sf = createFactory(isNetty()); - - // Make sure first producer grabs all the credits - sf.setProducerWindowSize(4000); - sf.setConsumerWindowSize(1024); - sf.setAckBatchSize(1024); - - final ClientSession session = sf.createSession(false, true, true, true); - - final SimpleString queueName = new SimpleString("testqueue"); - - session.createQueue(address, queueName, null, false); - - ClientProducer producer = session.createProducer(address); - - byte[] bytes = new byte[0]; - - ClientMessage message = session.createMessage(false); - - message.getBodyBuffer().writeBytes(bytes); - - producer.send(message); - - ClientSession session2 = sf.createSession(false, true, true, true); - - ClientProducer producer2 = session2.createProducer(address); - - ServerProducerCreditManager mgr = server.getPostOffice() - .getPagingManager() - .getPageStore(address) - .getProducerCreditManager(); - - long start = System.currentTimeMillis(); - - int waiting; - do - { - waiting = mgr.waitingEntries(); - - Thread.sleep(10); - } - while (waiting != 1 || System.currentTimeMillis() - start > 3000); - - Assert.assertEquals(1, waiting); - - ClientSession session3 = sf.createSession(false, true, true, true); - - ClientProducer producer3 = session3.createProducer(address); - - start = System.currentTimeMillis(); - - do - { - waiting = mgr.waitingEntries(); - - Thread.sleep(10); - } - while (waiting != 2 || System.currentTimeMillis() - start > 3000); - - Assert.assertEquals(2, waiting); - - session2.close(); - - session.close(); - - message = session.createMessage(false); - - message.getBodyBuffer().writeBytes(bytes); - - producer3.send(message); - - session3.close(); - - TestSupportPageStore store = (TestSupportPageStore)server.getPostOffice() - .getPagingManager() - .getPageStore(address); - - Assert.assertFalse(store.isExceededAvailableCredits()); - - server.stop(); - } - public void testClosingSessionUnblocksBlockedProducer() throws Exception { final SimpleString address = new SimpleString("testaddress"); @@ -690,12 +396,6 @@ t.join(); - TestSupportPageStore store = (TestSupportPageStore)server.getPostOffice() - .getPagingManager() - .getPageStore(address); - - Assert.assertFalse(store.isExceededAvailableCredits()); - server.stop(); } @@ -739,12 +439,6 @@ session.close(); - TestSupportPageStore store = (TestSupportPageStore)server.getPostOffice() - .getPagingManager() - .getPageStore(address); - - Assert.assertFalse(store.isExceededAvailableCredits()); - server.stop(); } @@ -827,31 +521,31 @@ final ClientSession session = sf.createSession(false, true, true, true); session.createQueue("address", "queue1", null, false); - + ClientProducerCredits credits = null; - + for (int i = 0; i < ClientProducerCreditManagerImpl.MAX_UNREFERENCED_CREDITS_CACHE_SIZE * 2; i++) { ClientProducer prod = session.createProducer("address"); - - ClientProducerCredits newCredits = ((ClientProducerInternal)prod).getProducerCredits(); - + + ClientProducerCredits newCredits = ((ClientProducerInternal)prod).getProducerCredits(); + if (credits != null) - { + { assertTrue(newCredits == credits); } - + credits = newCredits; - + assertEquals(1, ((ClientSessionInternal)session).getProducerCreditManager().creditsMapSize()); assertEquals(0, ((ClientSessionInternal)session).getProducerCreditManager().unReferencedCreditsSize()); } - + session.close(); server.stop(); } - + public void testProducerCreditsCaching2() throws Exception { HornetQServer server = createServer(false, isNetty()); @@ -863,33 +557,33 @@ final ClientSession session = sf.createSession(false, true, true, true); session.createQueue("address", "queue1", null, false); - + ClientProducerCredits credits = null; - + for (int i = 0; i < ClientProducerCreditManagerImpl.MAX_UNREFERENCED_CREDITS_CACHE_SIZE * 2; i++) { ClientProducer prod = session.createProducer("address"); - - ClientProducerCredits newCredits = ((ClientProducerInternal)prod).getProducerCredits(); - + + ClientProducerCredits newCredits = ((ClientProducerInternal)prod).getProducerCredits(); + if (credits != null) - { + { assertTrue(newCredits == credits); } - + credits = newCredits; - + prod.close(); - + assertEquals(1, ((ClientSessionInternal)session).getProducerCreditManager().creditsMapSize()); assertEquals(1, ((ClientSessionInternal)session).getProducerCreditManager().unReferencedCreditsSize()); } - + session.close(); server.stop(); } - + public void testProducerCreditsCaching3() throws Exception { HornetQServer server = createServer(false, isNetty()); @@ -901,31 +595,31 @@ final ClientSession session = sf.createSession(false, true, true, true); session.createQueue("address", "queue1", null, false); - + ClientProducerCredits credits = null; - + for (int i = 0; i < ClientProducerCreditManagerImpl.MAX_UNREFERENCED_CREDITS_CACHE_SIZE; i++) { ClientProducer prod = session.createProducer("address" + i); - - ClientProducerCredits newCredits = ((ClientProducerInternal)prod).getProducerCredits(); - + + ClientProducerCredits newCredits = ((ClientProducerInternal)prod).getProducerCredits(); + if (credits != null) - { + { assertFalse(newCredits == credits); } - + credits = newCredits; - + assertEquals(i + 1, ((ClientSessionInternal)session).getProducerCreditManager().creditsMapSize()); assertEquals(0, ((ClientSessionInternal)session).getProducerCreditManager().unReferencedCreditsSize()); } - + session.close(); server.stop(); } - + public void testProducerCreditsCaching4() throws Exception { HornetQServer server = createServer(false, isNetty()); @@ -937,33 +631,33 @@ final ClientSession session = sf.createSession(false, true, true, true); session.createQueue("address", "queue1", null, false); - + ClientProducerCredits credits = null; - + for (int i = 0; i < ClientProducerCreditManagerImpl.MAX_UNREFERENCED_CREDITS_CACHE_SIZE; i++) { ClientProducer prod = session.createProducer("address" + i); - - ClientProducerCredits newCredits = ((ClientProducerInternal)prod).getProducerCredits(); - + + ClientProducerCredits newCredits = ((ClientProducerInternal)prod).getProducerCredits(); + if (credits != null) - { + { assertFalse(newCredits == credits); } - + credits = newCredits; - + prod.close(); - + assertEquals(i + 1, ((ClientSessionInternal)session).getProducerCreditManager().creditsMapSize()); assertEquals(i + 1, ((ClientSessionInternal)session).getProducerCreditManager().unReferencedCreditsSize()); } - + session.close(); server.stop(); } - + public void testProducerCreditsCaching5() throws Exception { HornetQServer server = createServer(false, isNetty()); @@ -975,57 +669,59 @@ final ClientSession session = sf.createSession(false, true, true, true); session.createQueue("address", "queue1", null, false); - + ClientProducerCredits credits = null; - + List<ClientProducerCredits> creditsList = new ArrayList<ClientProducerCredits>(); - + for (int i = 0; i < ClientProducerCreditManagerImpl.MAX_UNREFERENCED_CREDITS_CACHE_SIZE; i++) { ClientProducer prod = session.createProducer("address" + i); - - ClientProducerCredits newCredits = ((ClientProducerInternal)prod).getProducerCredits(); - + + ClientProducerCredits newCredits = ((ClientProducerInternal)prod).getProducerCredits(); + if (credits != null) - { + { assertFalse(newCredits == credits); } - + credits = newCredits; - + assertEquals(i + 1, ((ClientSessionInternal)session).getProducerCreditManager().creditsMapSize()); assertEquals(0, ((ClientSessionInternal)session).getProducerCreditManager().unReferencedCreditsSize()); - + creditsList.add(credits); } - + Iterator<ClientProducerCredits> iter = creditsList.iterator(); - + for (int i = 0; i < ClientProducerCreditManagerImpl.MAX_UNREFERENCED_CREDITS_CACHE_SIZE; i++) { ClientProducer prod = session.createProducer("address" + i); - - ClientProducerCredits newCredits = ((ClientProducerInternal)prod).getProducerCredits(); - + + ClientProducerCredits newCredits = ((ClientProducerInternal)prod).getProducerCredits(); + assertTrue(newCredits == iter.next()); - - assertEquals(ClientProducerCreditManagerImpl.MAX_UNREFERENCED_CREDITS_CACHE_SIZE, ((ClientSessionInternal)session).getProducerCreditManager().creditsMapSize()); + + assertEquals(ClientProducerCreditManagerImpl.MAX_UNREFERENCED_CREDITS_CACHE_SIZE, + ((ClientSessionInternal)session).getProducerCreditManager().creditsMapSize()); assertEquals(0, ((ClientSessionInternal)session).getProducerCreditManager().unReferencedCreditsSize()); } - + for (int i = 0; i < 10; i++) { ClientProducer prod = session.createProducer("address" + (i + ClientProducerCreditManagerImpl.MAX_UNREFERENCED_CREDITS_CACHE_SIZE)); - - assertEquals(ClientProducerCreditManagerImpl.MAX_UNREFERENCED_CREDITS_CACHE_SIZE + i + 1, ((ClientSessionInternal)session).getProducerCreditManager().creditsMapSize()); + + assertEquals(ClientProducerCreditManagerImpl.MAX_UNREFERENCED_CREDITS_CACHE_SIZE + i + 1, + ((ClientSessionInternal)session).getProducerCreditManager().creditsMapSize()); assertEquals(0, ((ClientSessionInternal)session).getProducerCreditManager().unReferencedCreditsSize()); } - + session.close(); server.stop(); } - + public void testProducerCreditsCaching6() throws Exception { HornetQServer server = createServer(false, isNetty()); @@ -1037,22 +733,22 @@ final ClientSession session = sf.createSession(false, true, true, true); session.createQueue("address", "queue1", null, false); - + for (int i = 0; i < ClientProducerCreditManagerImpl.MAX_UNREFERENCED_CREDITS_CACHE_SIZE; i++) { ClientProducer prod = session.createProducer((String)null); - + prod.send("address", session.createMessage(false)); - + assertEquals(1, ((ClientSessionInternal)session).getProducerCreditManager().creditsMapSize()); assertEquals(1, ((ClientSessionInternal)session).getProducerCreditManager().unReferencedCreditsSize()); } - + session.close(); server.stop(); } - + public void testProducerCreditsCaching7() throws Exception { HornetQServer server = createServer(false, isNetty()); @@ -1064,42 +760,46 @@ final ClientSession session = sf.createSession(false, true, true, true); session.createQueue("address", "queue1", null, false); - + for (int i = 0; i < ClientProducerCreditManagerImpl.MAX_UNREFERENCED_CREDITS_CACHE_SIZE; i++) { ClientProducer prod = session.createProducer((String)null); - + prod.send("address" + i, session.createMessage(false)); - + assertEquals(i + 1, ((ClientSessionInternal)session).getProducerCreditManager().creditsMapSize()); assertEquals(i + 1, ((ClientSessionInternal)session).getProducerCreditManager().unReferencedCreditsSize()); } - + for (int i = 0; i < 10; i++) { ClientProducer prod = session.createProducer((String)null); - + prod.send("address" + i, session.createMessage(false)); - - assertEquals(ClientProducerCreditManagerImpl.MAX_UNREFERENCED_CREDITS_CACHE_SIZE, ((ClientSessionInternal)session).getProducerCreditManager().creditsMapSize()); - assertEquals(ClientProducerCreditManagerImpl.MAX_UNREFERENCED_CREDITS_CACHE_SIZE, ((ClientSessionInternal)session).getProducerCreditManager().unReferencedCreditsSize()); + + assertEquals(ClientProducerCreditManagerImpl.MAX_UNREFERENCED_CREDITS_CACHE_SIZE, + ((ClientSessionInternal)session).getProducerCreditManager().creditsMapSize()); + assertEquals(ClientProducerCreditManagerImpl.MAX_UNREFERENCED_CREDITS_CACHE_SIZE, + ((ClientSessionInternal)session).getProducerCreditManager().unReferencedCreditsSize()); } - + for (int i = 0; i < 10; i++) { ClientProducer prod = session.createProducer((String)null); - + prod.send("address2-" + i, session.createMessage(false)); - - assertEquals(ClientProducerCreditManagerImpl.MAX_UNREFERENCED_CREDITS_CACHE_SIZE, ((ClientSessionInternal)session).getProducerCreditManager().creditsMapSize()); - assertEquals(ClientProducerCreditManagerImpl.MAX_UNREFERENCED_CREDITS_CACHE_SIZE, ((ClientSessionInternal)session).getProducerCreditManager().unReferencedCreditsSize()); + + assertEquals(ClientProducerCreditManagerImpl.MAX_UNREFERENCED_CREDITS_CACHE_SIZE, + ((ClientSessionInternal)session).getProducerCreditManager().creditsMapSize()); + assertEquals(ClientProducerCreditManagerImpl.MAX_UNREFERENCED_CREDITS_CACHE_SIZE, + ((ClientSessionInternal)session).getProducerCreditManager().unReferencedCreditsSize()); } - + session.close(); server.stop(); } - + public void testProducerCreditsRefCounting() throws Exception { HornetQServer server = createServer(false, isNetty()); @@ -1111,37 +811,37 @@ final ClientSession session = sf.createSession(false, true, true, true); session.createQueue("address", "queue1", null, false); - + ClientProducer prod1 = session.createProducer("address"); assertEquals(1, ((ClientSessionInternal)session).getProducerCreditManager().creditsMapSize()); assertEquals(0, ((ClientSessionInternal)session).getProducerCreditManager().unReferencedCreditsSize()); - + ClientProducer prod2 = session.createProducer("address"); assertEquals(1, ((ClientSessionInternal)session).getProducerCreditManager().creditsMapSize()); assertEquals(0, ((ClientSessionInternal)session).getProducerCreditManager().unReferencedCreditsSize()); - + ClientProducer prod3 = session.createProducer("address"); assertEquals(1, ((ClientSessionInternal)session).getProducerCreditManager().creditsMapSize()); assertEquals(0, ((ClientSessionInternal)session).getProducerCreditManager().unReferencedCreditsSize()); - + prod1.close(); - + assertEquals(1, ((ClientSessionInternal)session).getProducerCreditManager().creditsMapSize()); assertEquals(0, ((ClientSessionInternal)session).getProducerCreditManager().unReferencedCreditsSize()); - + prod2.close(); - + assertEquals(1, ((ClientSessionInternal)session).getProducerCreditManager().creditsMapSize()); assertEquals(0, ((ClientSessionInternal)session).getProducerCreditManager().unReferencedCreditsSize()); - + prod3.close(); - + assertEquals(1, ((ClientSessionInternal)session).getProducerCreditManager().creditsMapSize()); assertEquals(1, ((ClientSessionInternal)session).getProducerCreditManager().unReferencedCreditsSize()); - + session.close(); server.stop(); } - + } Modified: trunk/tests/src/org/hornetq/tests/integration/client/ReceiveTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/client/ReceiveTest.java 2010-03-22 23:43:26 UTC (rev 8946) +++ trunk/tests/src/org/hornetq/tests/integration/client/ReceiveTest.java 2010-03-23 15:11:11 UTC (rev 8947) @@ -177,10 +177,13 @@ cp.send(sendSession.createMessage(false)); cp.send(sendSession.createMessage(false)); cp.send(sendSession.createMessage(false)); - // at this point we know that the first consumer has a messge in ites buffer + Assert.assertNotNull(cc2.receive(5000)); - Assert.assertNotNull(cc2.receive(5000)); - Assert.assertNotNull(cc.receiveImmediate()); + Assert.assertNotNull(cc.receive(5000)); + if (cc.receiveImmediate() == null) + { + assertNotNull(cc2.receiveImmediate()); + } session.close(); sendSession.close(); } Modified: trunk/tests/src/org/hornetq/tests/integration/cluster/failover/FailoverTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/cluster/failover/FailoverTest.java 2010-03-22 23:43:26 UTC (rev 8946) +++ trunk/tests/src/org/hornetq/tests/integration/cluster/failover/FailoverTest.java 2010-03-23 15:11:11 UTC (rev 8947) @@ -2270,6 +2270,7 @@ */ protected void setBody(final int i, final ClientMessage message) throws Exception { + log.info("in failovertest:: setbody"); message.getBodyBuffer().writeString("message" + i); } Modified: trunk/tests/src/org/hornetq/tests/integration/cluster/failover/LargeMessageFailoverTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/cluster/failover/LargeMessageFailoverTest.java 2010-03-22 23:43:26 UTC (rev 8946) +++ trunk/tests/src/org/hornetq/tests/integration/cluster/failover/LargeMessageFailoverTest.java 2010-03-23 15:11:11 UTC (rev 8947) @@ -14,11 +14,11 @@ package org.hornetq.tests.integration.cluster.failover; import junit.framework.Assert; -import junit.framework.TestSuite; import org.hornetq.api.core.HornetQBuffer; import org.hornetq.api.core.client.ClientMessage; import org.hornetq.api.core.client.HornetQClient; +import org.hornetq.core.logging.Logger; import org.hornetq.tests.util.UnitTestCase; /** @@ -33,40 +33,13 @@ // Constants ----------------------------------------------------- + private static final Logger log = Logger.getLogger(LargeMessageFailoverTest.class); + + // Attributes ---------------------------------------------------- // Static -------------------------------------------------------- - public static TestSuite suite() - { - TestSuite suite = new TestSuite(); - - suite.addTest(new LargeMessageFailoverTest("testNonTransacted")); - suite.addTest(new LargeMessageFailoverTest("testTransactedMessagesSentSoRollback")); - suite.addTest(new LargeMessageFailoverTest("testTransactedMessagesNotSentSoNoRollback")); - suite.addTest(new LargeMessageFailoverTest("testTransactedMessagesConsumedSoRollback")); - suite.addTest(new LargeMessageFailoverTest("testTransactedMessagesNotConsumedSoNoRollback")); - suite.addTest(new LargeMessageFailoverTest("testXAMessagesSentSoRollbackOnEnd")); - suite.addTest(new LargeMessageFailoverTest("testXAMessagesSentSoRollbackOnPrepare")); - suite.addTest(new LargeMessageFailoverTest("testXAMessagesSentSoRollbackOnCommit")); - suite.addTest(new LargeMessageFailoverTest("testXAMessagesNotSentSoNoRollbackOnCommit")); - suite.addTest(new LargeMessageFailoverTest("testXAMessagesConsumedSoRollbackOnEnd")); - suite.addTest(new LargeMessageFailoverTest("testXAMessagesConsumedSoRollbackOnPrepare")); - suite.addTest(new LargeMessageFailoverTest("testXAMessagesConsumedSoRollbackOnCommit")); - suite.addTest(new LargeMessageFailoverTest("testCreateNewFactoryAfterFailover")); - - // Those tests are temporarily disabled for LargeMessage - suite.addTest(new LargeMessageFailoverTest("testFailoverMultipleSessionsWithConsumers")); - suite.addTest(new LargeMessageFailoverTest("testFailWithBrowser")); - suite.addTest(new LargeMessageFailoverTest("testFailThenReceiveMoreMessagesAfterFailover")); - suite.addTest(new LargeMessageFailoverTest("testFailThenReceiveMoreMessagesAfterFailover2")); - - suite.addTest(new LargeMessageFailoverTest("testForceBlockingReturn")); - suite.addTest(new LargeMessageFailoverTest("testCommitOccurredUnblockedAndResendNoDuplicates")); - suite.addTest(new LargeMessageFailoverTest("testCommitDidNotOccurUnblockedAndResend")); - return suite; - } - // Constructors -------------------------------------------------- // Public -------------------------------------------------------- @@ -102,7 +75,7 @@ for (int j = 0; j < HornetQClient.DEFAULT_MIN_LARGE_MESSAGE_SIZE * 3; j++) { - Assert.assertEquals(buffer.readByte(), UnitTestCase.getSamplebyte(j)); + Assert.assertEquals("equal at " + j, buffer.readByte(), UnitTestCase.getSamplebyte(j)); } } Modified: trunk/tests/src/org/hornetq/tests/integration/divert/PersistentDivertTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/divert/PersistentDivertTest.java 2010-03-22 23:43:26 UTC (rev 8946) +++ trunk/tests/src/org/hornetq/tests/integration/divert/PersistentDivertTest.java 2010-03-23 15:11:11 UTC (rev 8947) @@ -20,7 +20,12 @@ import org.hornetq.api.core.SimpleString; import org.hornetq.api.core.TransportConfiguration; -import org.hornetq.api.core.client.*; +import org.hornetq.api.core.client.ClientConsumer; +import org.hornetq.api.core.client.ClientMessage; +import org.hornetq.api.core.client.ClientProducer; +import org.hornetq.api.core.client.ClientSession; +import org.hornetq.api.core.client.ClientSessionFactory; +import org.hornetq.api.core.client.HornetQClient; import org.hornetq.core.config.Configuration; import org.hornetq.core.config.DivertConfiguration; import org.hornetq.core.logging.Logger; @@ -481,8 +486,6 @@ sf.close(); - Assert.assertEquals(0, messagingService.getPostOffice().getPagingManager().getTotalMemory()); - messagingService.stop(); } Modified: trunk/tests/src/org/hornetq/tests/integration/jms/client/TextMessageTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/jms/client/TextMessageTest.java 2010-03-22 23:43:26 UTC (rev 8946) +++ trunk/tests/src/org/hornetq/tests/integration/jms/client/TextMessageTest.java 2010-03-23 15:11:11 UTC (rev 8947) @@ -173,17 +173,23 @@ TextMessage received2 = (TextMessage)cons.receive(1000); Assert.assertNotNull(received2); Assert.assertEquals(str, received2.getText()); + + assertEquals(str, msg2.getText()); // Now resend it prod.send(received2); + assertEquals(str, received2.getText()); TextMessage received3 = (TextMessage)cons.receive(1000); Assert.assertNotNull(received3); + assertEquals(str, received3.getText()); // And resend again prod.send(received3); + assertEquals(str, received3.getText()); TextMessage received4 = (TextMessage)cons.receive(1000); Assert.assertNotNull(received4); + assertEquals(str, received4.getText()); } finally Modified: trunk/tests/src/org/hornetq/tests/integration/jms/server/management/JMSQueueControlTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/jms/server/management/JMSQueueControlTest.java 2010-03-22 23:43:26 UTC (rev 8946) +++ trunk/tests/src/org/hornetq/tests/integration/jms/server/management/JMSQueueControlTest.java 2010-03-23 15:11:11 UTC (rev 8947) @@ -823,6 +823,7 @@ conf.setSecurityEnabled(false); conf.setJMXManagementEnabled(true); conf.getAcceptorConfigurations().add(new TransportConfiguration(InVMAcceptorFactory.class.getName())); + conf.setFileDeploymentEnabled(false); server = HornetQServers.newHornetQServer(conf, mbeanServer, false); server.start(); Modified: trunk/tests/src/org/hornetq/tests/integration/jms/server/management/JMSQueueControlUsingJMSTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/jms/server/management/JMSQueueControlUsingJMSTest.java 2010-03-22 23:43:26 UTC (rev 8946) +++ trunk/tests/src/org/hornetq/tests/integration/jms/server/management/JMSQueueControlUsingJMSTest.java 2010-03-23 15:11:11 UTC (rev 8947) @@ -135,7 +135,7 @@ return (Integer)proxy.retrieveAttributeValue("messageCount"); } - public int getMessagesAdded() + public long getMessagesAdded() { return (Integer)proxy.retrieveAttributeValue("messagesAdded"); } Modified: trunk/tests/src/org/hornetq/tests/integration/journal/JournalPerfTuneTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/journal/JournalPerfTuneTest.java 2010-03-22 23:43:26 UTC (rev 8946) +++ trunk/tests/src/org/hornetq/tests/integration/journal/JournalPerfTuneTest.java 2010-03-23 15:11:11 UTC (rev 8947) @@ -30,7 +30,6 @@ import org.hornetq.core.journal.impl.AIOSequentialFileFactory; import org.hornetq.core.journal.impl.JournalImpl; import org.hornetq.core.logging.Logger; -import org.hornetq.tests.integration.cluster.failover.LargeMessageFailoverTest; import org.hornetq.tests.util.UnitTestCase; import org.hornetq.utils.DataConstants; @@ -144,6 +143,8 @@ public void done() { latch.countDown(); + + log.info(latch.getCount()); } public void onError(int errorCode, String errorMessage) @@ -202,7 +203,7 @@ { try { - Record record = new Record(new byte[256]); + Record record = new Record(new byte[1024]); for (int i = 0; i < iters; i++) { Modified: trunk/tests/src/org/hornetq/tests/integration/largemessage/LargeMessageTestBase.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/largemessage/LargeMessageTestBase.java 2010-03-22 23:43:26 UTC (rev 8946) +++ trunk/tests/src/org/hornetq/tests/integration/largemessage/LargeMessageTestBase.java 2010-03-23 15:11:11 UTC (rev 8947) @@ -523,8 +523,6 @@ session.close(); - long globalSize = server.getPostOffice().getPagingManager().getTotalMemory(); - Assert.assertEquals(0l, globalSize); Assert.assertEquals(0, ((Queue)server.getPostOffice().getBinding(ADDRESS).getBindable()).getDeliveringCount()); Assert.assertEquals(0, ((Queue)server.getPostOffice().getBinding(ADDRESS).getBindable()).getMessageCount()); Modified: trunk/tests/src/org/hornetq/tests/integration/management/QueueControlUsingCoreTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/management/QueueControlUsingCoreTest.java 2010-03-22 23:43:26 UTC (rev 8946) +++ trunk/tests/src/org/hornetq/tests/integration/management/QueueControlUsingCoreTest.java 2010-03-23 15:11:11 UTC (rev 8947) @@ -22,7 +22,6 @@ import org.hornetq.api.core.client.HornetQClient; import org.hornetq.api.core.management.QueueControl; import org.hornetq.api.core.management.ResourceNames; -import org.hornetq.core.client.impl.ClientSessionFactoryImpl; import org.hornetq.core.logging.Logger; import org.hornetq.core.remoting.impl.invm.InVMConnectorFactory; @@ -116,7 +115,7 @@ return (Integer)proxy.retrieveAttributeValue("messageCount"); } - public int getMessagesAdded() + public long getMessagesAdded() { return (Integer)proxy.retrieveAttributeValue("messagesAdded"); } Modified: trunk/tests/src/org/hornetq/tests/integration/paging/PageCrashTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/paging/PageCrashTest.java 2010-03-22 23:43:26 UTC (rev 8946) +++ trunk/tests/src/org/hornetq/tests/integration/paging/PageCrashTest.java 2010-03-23 15:11:11 UTC (rev 8947) @@ -189,8 +189,6 @@ session.close(); - Assert.assertTrue(server.getPostOffice().getPagingManager().getTotalMemory() > 0); - session = sf.createSession(null, null, false, true, true, false, 0); ClientConsumer consumer = session.createConsumer(PageCrashTest.ADDRESS); @@ -209,9 +207,6 @@ consumer.close(); session.close(); - - Assert.assertEquals(0, server.getPostOffice().getPagingManager().getTotalMemory()); - } finally { Modified: trunk/tests/src/org/hornetq/tests/integration/paging/PagingSendTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/paging/PagingSendTest.java 2010-03-22 23:43:26 UTC (rev 8946) +++ trunk/tests/src/org/hornetq/tests/integration/paging/PagingSendTest.java 2010-03-23 15:11:11 UTC (rev 8947) @@ -119,8 +119,6 @@ session.close(); - Assert.assertTrue(server.getPostOffice().getPagingManager().getTotalMemory() > 0); - session = sf.createSession(null, null, false, true, true, false, 0); ClientConsumer consumer = session.createConsumer(PagingSendTest.ADDRESS); @@ -144,9 +142,6 @@ consumer.close(); session.close(); - - Assert.assertEquals(0, server.getPostOffice().getPagingManager().getTotalMemory()); - } finally { Deleted: trunk/tests/src/org/hornetq/tests/integration/remoting/AardvarkProtocolTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/remoting/AardvarkProtocolTest.java 2010-03-22 23:43:26 UTC (rev 8946) +++ trunk/tests/src/org/hornetq/tests/integration/remoting/AardvarkProtocolTest.java 2010-03-23 15:11:11 UTC (rev 8947) @@ -1,98 +0,0 @@ -/* - * Copyright 2010 Red Hat, Inc. - * Red Hat licenses this file to you under the Apache License, version - * 2.0 (the "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * http://www.apache.org/licenses/LICENSE-2.0 - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or - * implied. See the License for the specific language governing - * permissions and limitations under the License. - */ - -package org.hornetq.tests.integration.remoting; - -import java.io.BufferedOutputStream; -import java.io.InputStream; -import java.io.OutputStream; -import java.net.Socket; -import java.util.HashMap; -import java.util.Map; - -import org.hornetq.api.core.TransportConfiguration; -import org.hornetq.core.config.Configuration; -import org.hornetq.core.config.impl.ConfigurationImpl; -import org.hornetq.core.logging.Logger; -import org.hornetq.core.remoting.server.impl.RemotingServiceImpl; -import org.hornetq.core.server.HornetQServer; -import org.hornetq.core.server.HornetQServers; -import org.hornetq.integration.transports.netty.NettyAcceptorFactory; -import org.hornetq.integration.transports.netty.TransportConstants; -import org.hornetq.spi.core.protocol.ProtocolType; -import org.hornetq.tests.util.ServiceTestBase; - -/** - * An AardvarkProtocolTest - * - * @author Tim Fox - * - * - */ -public class AardvarkProtocolTest extends ServiceTestBase -{ - private static final Logger log = Logger.getLogger(AardvarkProtocolTest.class); - - public void testAardvark() throws Exception - { - Configuration config = new ConfigurationImpl(); - - config.setSecurityEnabled(false); - config.setPersistenceEnabled(false); - - Map<String, Object> params = new HashMap<String, Object>(); - - params.put(TransportConstants.PORT_PROP_NAME, 9876); - params.put(TransportConstants.HOST_PROP_NAME, "127.0.0.1"); - params.put(TransportConstants.PROTOCOL_PROP_NAME, ProtocolType.AARDVARK.toString()); - - TransportConfiguration tc = new TransportConfiguration(NettyAcceptorFactory.class.getCanonicalName(), - params); - - config.getAcceptorConfigurations().add(tc); - - HornetQServer server = HornetQServers.newHornetQServer(config); - - server.start(); - - //Now we should be able to make a connection to this port and talk the Aardvark protocol! - - Socket socket = new Socket("127.0.0.1", 9876); - - - - OutputStream out = new BufferedOutputStream(socket.getOutputStream()); - - String s = "AARDVARK!\n"; - - byte[] bytes = s.getBytes("UTF-8"); - - out.write(bytes); - - out.flush(); - - InputStream in = socket.getInputStream(); - - log.info("writing bytes"); - byte b; - - while ((b = (byte)in.read()) != '\n') - { - log.info("read " + (char)b); - } - - socket.close(); - - server.stop(); - } -} Modified: trunk/tests/src/org/hornetq/tests/integration/security/NettySecurityClientTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/security/NettySecurityClientTest.java 2010-03-22 23:43:26 UTC (rev 8946) +++ trunk/tests/src/org/hornetq/tests/integration/security/NettySecurityClientTest.java 2010-03-23 15:11:11 UTC (rev 8947) @@ -124,7 +124,8 @@ } else { - Assert.fail("Exception when starting the client: " + line); + //Assert.fail("Exception when starting the client: " + line); + System.out.println(line); } } Modified: trunk/tests/src/org/hornetq/tests/integration/security/SimpleClient.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/security/SimpleClient.java 2010-03-22 23:43:26 UTC (rev 8946) +++ trunk/tests/src/org/hornetq/tests/integration/security/SimpleClient.java 2010-03-23 15:11:11 UTC (rev 8947) @@ -85,12 +85,15 @@ } catch (Throwable t) { + String allStack = t.getMessage() + "|"; StackTraceElement[] stackTrace = t.getStackTrace(); for (StackTraceElement stackTraceElement : stackTrace) { allStack += stackTraceElement.toString() + "|"; } + //System.out.println(t.getClass().getName()); + //System.out.println(t.getMessage()); System.out.println(allStack); System.exit(1); } Modified: trunk/tests/src/org/hornetq/tests/integration/xa/BasicXaRecoveryTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/xa/BasicXaRecoveryTest.java 2010-03-22 23:43:26 UTC (rev 8946) +++ trunk/tests/src/org/hornetq/tests/integration/xa/BasicXaRecoveryTest.java 2010-03-23 15:11:11 UTC (rev 8947) @@ -344,8 +344,6 @@ clientSession.createQueue(pageQueue, pageQueue, null, true); - long initialPageSize = server.getPostOffice().getPagingManager().getTotalMemory(); - clientSession.start(xid, XAResource.TMNOFLAGS); ClientProducer pageProducer = clientSession.createProducer(pageQueue); @@ -382,13 +380,8 @@ Assert.assertNull(pageConsumer.receiveImmediate()); - long globalSize = server.getPostOffice().getPagingManager().getTotalMemory(); // Management message (from createQueue) will not be taken into account again as it is nonPersistent - BasicXaRecoveryTest.log.info("global size is " + globalSize + " initial page size is " + initialPageSize); - - Assert.assertTrue(globalSize == initialPageSize || globalSize == 0l); - } public void testNonPersistent() throws Exception Deleted: trunk/tests/src/org/hornetq/tests/unit/core/list/impl/PriorityLinkedListTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/unit/core/list/impl/PriorityLinkedListTest.java 2010-03-22 23:43:26 UTC (rev 8946) +++ trunk/tests/src/org/hornetq/tests/unit/core/list/impl/PriorityLinkedListTest.java 2010-03-23 15:11:11 UTC (rev 8947) @@ -1,659 +0,0 @@ -/* - * Copyright 2009 Red Hat, Inc. - * Red Hat licenses this file to you under the Apache License, version - * 2.0 (the "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * http://www.apache.org/licenses/LICENSE-2.0 - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or - * implied. See the License for the specific language governing - * permissions and limitations under the License. - */ - -package org.hornetq.tests.unit.core.list.impl; - -import java.util.Iterator; - -import junit.framework.Assert; - -import org.hornetq.core.list.PriorityLinkedList; -import org.hornetq.core.list.impl.PriorityLinkedListImpl; -import org.hornetq.tests.util.UnitTestCase; - -/** - * @author <a href=&quot;tim.fox(a)jboss.com&gt;Tim Fox</a> - * - * $Id$ - */ -public class PriorityLinkedListTest extends UnitTestCase -{ - protected PriorityLinkedList<Wibble> list; - - protected Wibble a; - - protected Wibble b; - - protected Wibble c; - - protected Wibble d; - - protected Wibble e; - - protected Wibble f; - - protected Wibble g; - - protected Wibble h; - - protected Wibble i; - - protected Wibble j; - - protected Wibble k; - - protected Wibble l; - - protected Wibble m; - - protected Wibble n; - - protected Wibble o; - - protected Wibble p; - - protected Wibble q; - - protected Wibble r; - - protected Wibble s; - - protected Wibble t; - - protected Wibble u; - - protected Wibble v; - - protected Wibble w; - - protected Wibble x; - - protected Wibble y; - - protected Wibble z; - - public void setUp() throws Exception - { - super.setUp(); - - list = new PriorityLinkedListImpl<Wibble>(10); - - a = new Wibble("a"); - b = new Wibble("b"); - c = new Wibble("c"); - d = new Wibble("d"); - e = new Wibble("e"); - f = new Wibble("f"); - g = new Wibble("g"); - h = new Wibble("h"); - i = new Wibble("i"); - j = new Wibble("j"); - k = new Wibble("k"); - l = new Wibble("l"); - m = new Wibble("m"); - n = new Wibble("n"); - o = new Wibble("o"); - p = new Wibble("p"); - q = new Wibble("q"); - r = new Wibble("r"); - s = new Wibble("s"); - t = new Wibble("t"); - u = new Wibble("u"); - v = new Wibble("v"); - w = new Wibble("w"); - x = new Wibble("x"); - y = new Wibble("y"); - z = new Wibble("z"); - } - - public void tearDown() throws Exception - { - list = null; - super.tearDown(); - } - - public void testEmpty() throws Exception - { - Assert.assertTrue(list.isEmpty()); - - list.addFirst(a, 0); - - Assert.assertFalse(list.isEmpty()); - - Wibble w = list.removeFirst(); - Assert.assertEquals(a, w); - Assert.assertTrue(list.isEmpty()); - } - - public void testAddFirst() throws Exception - { - list.addFirst(a, 0); - list.addFirst(b, 0); - list.addFirst(c, 0); - list.addFirst(d, 0); - list.addFirst(e, 0); - - Assert.assertEquals(e, list.removeFirst()); - Assert.assertEquals(d, list.removeFirst()); - Assert.assertEquals(c, list.removeFirst()); - Assert.assertEquals(b, list.removeFirst()); - Assert.assertEquals(a, list.removeFirst()); - Assert.assertNull(list.removeFirst()); - } - - public void testAddLast() throws Exception - { - list.addLast(a, 0); - list.addLast(b, 0); - list.addLast(c, 0); - list.addLast(d, 0); - list.addLast(e, 0); - - Assert.assertEquals(a, list.removeFirst()); - Assert.assertEquals(b, list.removeFirst()); - Assert.assertEquals(c, list.removeFirst()); - Assert.assertEquals(d, list.removeFirst()); - Assert.assertEquals(e, list.removeFirst()); - Assert.assertNull(list.removeFirst()); - - } - - public void testPeekFirst() - { - list.addLast(a, 0); - list.addLast(b, 1); - list.addLast(c, 2); - list.addLast(d, 3); - list.addLast(e, 4); - list.addLast(f, 5); - list.addLast(g, 6); - list.addLast(h, 7); - list.addLast(i, 8); - list.addLast(j, 9); - - Assert.assertEquals(j, list.peekFirst()); - Assert.assertEquals(j, list.peekFirst()); - - list.removeFirst(); - - Assert.assertEquals(i, list.peekFirst()); - Assert.assertEquals(i, list.peekFirst()); - - list.clear(); - } - - public void testRemoveFirst() throws Exception - { - list.addLast(a, 0); - list.addLast(b, 1); - list.addLast(c, 2); - list.addLast(d, 3); - list.addLast(e, 4); - list.addLast(f, 5); - list.addLast(g, 6); - list.addLast(h, 7); - list.addLast(i, 8); - list.addLast(j, 9); - - Assert.assertEquals(j, list.removeFirst()); - Assert.assertEquals(i, list.removeFirst()); - Assert.assertEquals(h, list.removeFirst()); - Assert.assertEquals(g, list.removeFirst()); - Assert.assertEquals(f, list.removeFirst()); - Assert.assertEquals(e, list.removeFirst()); - Assert.assertEquals(d, list.removeFirst()); - Assert.assertEquals(c, list.removeFirst()); - Assert.assertEquals(b, list.removeFirst()); - Assert.assertEquals(a, list.removeFirst()); - - Assert.assertNull(list.removeFirst()); - - list.addLast(a, 9); - list.addLast(b, 8); - list.addLast(c, 7); - list.addLast(d, 6); - list.addLast(e, 5); - list.addLast(f, 4); - list.addLast(g, 3); - list.addLast(h, 2); - list.addLast(i, 1); - list.addLast(j, 0); - - Assert.assertEquals(a, list.removeFirst()); - Assert.assertEquals(b, list.removeFirst()); - Assert.assertEquals(c, list.removeFirst()); - Assert.assertEquals(d, list.removeFirst()); - Assert.assertEquals(e, list.removeFirst()); - Assert.assertEquals(f, list.removeFirst()); - Assert.assertEquals(g, list.removeFirst()); - Assert.assertEquals(h, list.removeFirst()); - Assert.assertEquals(i, list.removeFirst()); - Assert.assertEquals(j, list.removeFirst()); - - Assert.assertNull(list.removeFirst()); - - list.addLast(a, 9); - list.addLast(b, 0); - list.addLast(c, 8); - list.addLast(d, 1); - list.addLast(e, 7); - list.addLast(f, 2); - list.addLast(g, 6); - list.addLast(h, 3); - list.addLast(i, 5); - list.addLast(j, 4); - - Assert.assertEquals(a, list.removeFirst()); - Assert.assertEquals(c, list.removeFirst()); - Assert.assertEquals(e, list.removeFirst()); - Assert.assertEquals(g, list.removeFirst()); - Assert.assertEquals(i, list.removeFirst()); - Assert.assertEquals(j, list.removeFirst()); - Assert.assertEquals(h, list.removeFirst()); - Assert.assertEquals(f, list.removeFirst()); - Assert.assertEquals(d, list.removeFirst()); - Assert.assertEquals(b, list.removeFirst()); - - Assert.assertNull(list.removeFirst()); - - list.addLast(a, 0); - list.addLast(b, 3); - list.addLast(c, 3); - list.addLast(d, 3); - list.addLast(e, 6); - list.addLast(f, 6); - list.addLast(g, 6); - list.addLast(h, 9); - list.addLast(i, 9); - list.addLast(j, 9); - - Assert.assertEquals(h, list.removeFirst()); - Assert.assertEquals(i, list.removeFirst()); - Assert.assertEquals(j, list.removeFirst()); - Assert.assertEquals(e, list.removeFirst()); - Assert.assertEquals(f, list.removeFirst()); - Assert.assertEquals(g, list.removeFirst()); - Assert.assertEquals(b, list.removeFirst()); - Assert.assertEquals(c, list.removeFirst()); - Assert.assertEquals(d, list.removeFirst()); - Assert.assertEquals(a, list.removeFirst()); - - Assert.assertNull(list.removeFirst()); - - list.addLast(a, 5); - list.addLast(b, 5); - list.addLast(c, 5); - list.addLast(d, 5); - list.addLast(e, 5); - list.addLast(f, 5); - list.addLast(g, 5); - list.addLast(h, 5); - list.addLast(i, 5); - list.addLast(j, 5); - - Assert.assertEquals(a, list.removeFirst()); - Assert.assertEquals(b, list.removeFirst()); - Assert.assertEquals(c, list.removeFirst()); - Assert.assertEquals(d, list.removeFirst()); - Assert.assertEquals(e, list.removeFirst()); - Assert.assertEquals(f, list.removeFirst()); - Assert.assertEquals(g, list.removeFirst()); - Assert.assertEquals(h, list.removeFirst()); - Assert.assertEquals(i, list.removeFirst()); - Assert.assertEquals(j, list.removeFirst()); - - Assert.assertNull(list.removeFirst()); - - list.addLast(j, 5); - list.addLast(i, 5); - list.addLast(h, 5); - list.addLast(g, 5); - list.addLast(f, 5); - list.addLast(e, 5); - list.addLast(d, 5); - list.addLast(c, 5); - list.addLast(b, 5); - list.addLast(a, 5); - - Assert.assertEquals(j, list.removeFirst()); - Assert.assertEquals(i, list.removeFirst()); - Assert.assertEquals(h, list.removeFirst()); - Assert.assertEquals(g, list.removeFirst()); - Assert.assertEquals(f, list.removeFirst()); - Assert.assertEquals(e, list.removeFirst()); - Assert.assertEquals(d, list.removeFirst()); - Assert.assertEquals(c, list.removeFirst()); - Assert.assertEquals(b, list.removeFirst()); - Assert.assertEquals(a, list.removeFirst()); - - Assert.assertNull(list.removeFirst()); - - } - - public void testGetAll() throws Exception - { - list.addLast(a, 0); - list.addLast(b, 3); - list.addLast(c, 3); - list.addLast(d, 3); - list.addLast(e, 6); - list.addLast(f, 6); - list.addLast(g, 6); - list.addLast(h, 9); - list.addLast(i, 9); - list.addLast(j, 9); - - Iterator<Wibble> iter = list.getAll().iterator(); - int count = 0; - while (iter.hasNext()) - { - Object o = iter.next(); - if (count == 0) - { - Assert.assertEquals(h, o); - } - if (count == 1) - { - Assert.assertEquals(i, o); - } - if (count == 2) - { - Assert.assertEquals(j, o); - } - if (count == 3) - { - Assert.assertEquals(e, o); - } - if (count == 4) - { - Assert.assertEquals(f, o); - } - if (count == 5) - { - Assert.assertEquals(g, o); - } - if (count == 6) - { - Assert.assertEquals(b, o); - } - if (count == 7) - { - Assert.assertEquals(c, o); - } - if (count == 8) - { - Assert.assertEquals(d, o); - } - if (count == 9) - { - Assert.assertEquals(a, o); - } - count++; - } - Assert.assertEquals(10, count); - } - - public void testIterator() - { - list.addLast(a, 9); - list.addLast(b, 9); - list.addLast(c, 8); - list.addLast(d, 8); - list.addLast(e, 7); - list.addLast(f, 7); - list.addLast(g, 7); - list.addLast(h, 6); - list.addLast(i, 6); - list.addLast(j, 6); - list.addLast(k, 5); - list.addLast(l, 5); - list.addLast(m, 4); - list.addLast(n, 4); - list.addLast(o, 4); - list.addLast(p, 3); - list.addLast(q, 3); - list.addLast(r, 3); - list.addLast(s, 2); - list.addLast(t, 2); - list.addLast(u, 2); - list.addLast(v, 1); - list.addLast(w, 1); - list.addLast(x, 1); - list.addLast(y, 0); - list.addLast(z, 0); - - Iterator<Wibble> iter = list.iterator(); - - int c = 0; - while (iter.hasNext()) - { - iter.next(); - c++; - } - Assert.assertEquals(c, 26); - Assert.assertEquals(26, list.size()); - - iter = list.iterator(); - Assert.assertTrue(iter.hasNext()); - Wibble w = (Wibble)iter.next(); - Assert.assertEquals("a", w.s); - w = (Wibble)iter.next(); - Assert.assertEquals("b", w.s); - w = (Wibble)iter.next(); - Assert.assertEquals("c", w.s); - w = (Wibble)iter.next(); - Assert.assertEquals("d", w.s); - w = (Wibble)iter.next(); - Assert.assertEquals("e", w.s); - w = (Wibble)iter.next(); - Assert.assertEquals("f", w.s); - w = (Wibble)iter.next(); - Assert.assertEquals("g", w.s); - w = (Wibble)iter.next(); - Assert.assertEquals("h", w.s); - w = (Wibble)iter.next(); - Assert.assertEquals("i", w.s); - w = (Wibble)iter.next(); - Assert.assertEquals("j", w.s); - w = (Wibble)iter.next(); - Assert.assertEquals("k", w.s); - w = (Wibble)iter.next(); - Assert.assertEquals("l", w.s); - w = (Wibble)iter.next(); - Assert.assertEquals("m", w.s); - w = (Wibble)iter.next(); - Assert.assertEquals("n", w.s); - w = (Wibble)iter.next(); - Assert.assertEquals("o", w.s); - w = (Wibble)iter.next(); - Assert.assertEquals("p", w.s); - w = (Wibble)iter.next(); - Assert.assertEquals("q", w.s); - w = (Wibble)iter.next(); - Assert.assertEquals("r", w.s); - w = (Wibble)iter.next(); - Assert.assertEquals("s", w.s); - w = (Wibble)iter.next(); - Assert.assertEquals("t", w.s); - w = (Wibble)iter.next(); - Assert.assertEquals("u", w.s); - w = (Wibble)iter.next(); - Assert.assertEquals("v", w.s); - w = (Wibble)iter.next(); - Assert.assertEquals("w", w.s); - w = (Wibble)iter.next(); - Assert.assertEquals("x", w.s); - w = (Wibble)iter.next(); - Assert.assertEquals("y", w.s); - w = (Wibble)iter.next(); - Assert.assertEquals("z", w.s); - Assert.assertFalse(iter.hasNext()); - - iter = list.iterator(); - Assert.assertTrue(iter.hasNext()); - w = (Wibble)iter.next(); - Assert.assertEquals("a", w.s); - - iter.remove(); - - Assert.assertEquals(25, list.size()); - - w = (Wibble)iter.next(); - Assert.assertEquals("b", w.s); - w = (Wibble)iter.next(); - Assert.assertEquals("c", w.s); - w = (Wibble)iter.next(); - Assert.assertEquals("d", w.s); - - iter.remove(); - - Assert.assertEquals(24, list.size()); - - w = (Wibble)iter.next(); - Assert.assertEquals("e", w.s); - w = (Wibble)iter.next(); - Assert.assertEquals("f", w.s); - w = (Wibble)iter.next(); - Assert.assertEquals("g", w.s); - w = (Wibble)iter.next(); - Assert.assertEquals("h", w.s); - w = (Wibble)iter.next(); - Assert.assertEquals("i", w.s); - w = (Wibble)iter.next(); - Assert.assertEquals("j", w.s); - - iter.remove(); - - Assert.assertEquals(23, list.size()); - - w = (Wibble)iter.next(); - Assert.assertEquals("k", w.s); - w = (Wibble)iter.next(); - Assert.assertEquals("l", w.s); - w = (Wibble)iter.next(); - Assert.assertEquals("m", w.s); - w = (Wibble)iter.next(); - Assert.assertEquals("n", w.s); - w = (Wibble)iter.next(); - Assert.assertEquals("o", w.s); - w = (Wibble)iter.next(); - Assert.assertEquals("p", w.s); - w = (Wibble)iter.next(); - Assert.assertEquals("q", w.s); - w = (Wibble)iter.next(); - Assert.assertEquals("r", w.s); - w = (Wibble)iter.next(); - Assert.assertEquals("s", w.s); - w = (Wibble)iter.next(); - Assert.assertEquals("t", w.s); - w = (Wibble)iter.next(); - Assert.assertEquals("u", w.s); - w = (Wibble)iter.next(); - Assert.assertEquals("v", w.s); - w = (Wibble)iter.next(); - Assert.assertEquals("w", w.s); - w = (Wibble)iter.next(); - Assert.assertEquals("x", w.s); - w = (Wibble)iter.next(); - Assert.assertEquals("y", w.s); - w = (Wibble)iter.next(); - Assert.assertEquals("z", w.s); - iter.remove(); - Assert.assertFalse(iter.hasNext()); - - iter = list.iterator(); - Assert.assertTrue(iter.hasNext()); - w = (Wibble)iter.next(); - Assert.assertEquals("b", w.s); - w = (Wibble)iter.next(); - Assert.assertEquals("c", w.s); - w = (Wibble)iter.next(); - Assert.assertEquals("e", w.s); - w = (Wibble)iter.next(); - Assert.assertEquals("f", w.s); - w = (Wibble)iter.next(); - Assert.assertEquals("g", w.s); - w = (Wibble)iter.next(); - Assert.assertEquals("h", w.s); - w = (Wibble)iter.next(); - Assert.assertEquals("i", w.s); - w = (Wibble)iter.next(); - Assert.assertEquals("k", w.s); - w = (Wibble)iter.next(); - Assert.assertEquals("l", w.s); - w = (Wibble)iter.next(); - Assert.assertEquals("m", w.s); - w = (Wibble)iter.next(); - Assert.assertEquals("n", w.s); - w = (Wibble)iter.next(); - Assert.assertEquals("o", w.s); - w = (Wibble)iter.next(); - Assert.assertEquals("p", w.s); - w = (Wibble)iter.next(); - Assert.assertEquals("q", w.s); - w = (Wibble)iter.next(); - Assert.assertEquals("r", w.s); - w = (Wibble)iter.next(); - Assert.assertEquals("s", w.s); - w = (Wibble)iter.next(); - Assert.assertEquals("t", w.s); - w = (Wibble)iter.next(); - Assert.assertEquals("u", w.s); - w = (Wibble)iter.next(); - Assert.assertEquals("v", w.s); - w = (Wibble)iter.next(); - Assert.assertEquals("w", w.s); - w = (Wibble)iter.next(); - Assert.assertEquals("x", w.s); - w = (Wibble)iter.next(); - Assert.assertEquals("y", w.s); - Assert.assertFalse(iter.hasNext()); - - } - - public void testClear() - { - list.addLast(a, 0); - list.addLast(b, 3); - list.addLast(c, 3); - list.addLast(d, 3); - list.addLast(e, 6); - list.addLast(f, 6); - list.addLast(g, 6); - list.addLast(h, 9); - list.addLast(i, 9); - list.addLast(j, 9); - - list.clear(); - - Assert.assertNull(list.removeFirst()); - - Assert.assertTrue(list.getAll().isEmpty()); - } - - class Wibble - { - String s; - - Wibble(final String s) - { - this.s = s; - } - - public String toString() - { - return s; - } - } - -} Copied: trunk/tests/src/org/hornetq/tests/unit/core/list/impl/PriorityLinkedListTestBase.java (from rev 8938, trunk/tests/src/org/hornetq/tests/unit/core/list/impl/PriorityLinkedListTest.java) =================================================================== --- trunk/tests/src/org/hornetq/tests/unit/core/list/impl/PriorityLinkedListTestBase.java (rev 0) +++ trunk/tests/src/org/hornetq/tests/unit/core/list/impl/PriorityLinkedListTestBase.java 2010-03-23 15:11:11 UTC (rev 8947) @@ -0,0 +1,692 @@ +/* + * Copyright 2009 Red Hat, Inc. + * Red Hat licenses this file to you under the Apache License, version + * 2.0 (the "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * http://www.apache.org/licenses/LICENSE-2.0 + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or + * implied. See the License for the specific language governing + * permissions and limitations under the License. + */ + +package org.hornetq.tests.unit.core.list.impl; + +import junit.framework.Assert; +import junit.framework.TestCase; + +import org.hornetq.utils.PriorityLinkedListImpl; +import org.hornetq.utils.concurrent.HQIterator; + +/** + * @author <a href=&quot;tim.fox(a)jboss.com&gt;Tim Fox</a> + * + * $Id$ + */ +public abstract class PriorityLinkedListTestBase extends TestCase +{ + protected Wibble a; + + protected Wibble b; + + protected Wibble c; + + protected Wibble d; + + protected Wibble e; + + protected Wibble f; + + protected Wibble g; + + protected Wibble h; + + protected Wibble i; + + protected Wibble j; + + protected Wibble k; + + protected Wibble l; + + protected Wibble m; + + protected Wibble n; + + protected Wibble o; + + protected Wibble p; + + protected Wibble q; + + protected Wibble r; + + protected Wibble s; + + protected Wibble t; + + protected Wibble u; + + protected Wibble v; + + protected Wibble w; + + protected Wibble x; + + protected Wibble y; + + protected Wibble z; + + private PriorityLinkedListImpl<Wibble> list; + + protected abstract PriorityLinkedListImpl<Wibble> getList(); + + public void setUp() throws Exception + { + super.setUp(); + + list = getList(); + + a = new Wibble("a"); + b = new Wibble("b"); + c = new Wibble("c"); + d = new Wibble("d"); + e = new Wibble("e"); + f = new Wibble("f"); + g = new Wibble("g"); + h = new Wibble("h"); + i = new Wibble("i"); + j = new Wibble("j"); + k = new Wibble("k"); + l = new Wibble("l"); + m = new Wibble("m"); + n = new Wibble("n"); + o = new Wibble("o"); + p = new Wibble("p"); + q = new Wibble("q"); + r = new Wibble("r"); + s = new Wibble("s"); + t = new Wibble("t"); + u = new Wibble("u"); + v = new Wibble("v"); + w = new Wibble("w"); + x = new Wibble("x"); + y = new Wibble("y"); + z = new Wibble("z"); + } + + public void tearDown() throws Exception + { + list = null; + super.tearDown(); + } + + public void testEmpty() throws Exception + { + Assert.assertTrue(list.isEmpty()); + + list.addFirst(a, 0); + + Assert.assertFalse(list.isEmpty()); + + Wibble w = list.removeFirst(); + Assert.assertEquals(a, w); + Assert.assertTrue(list.isEmpty()); + } + + public void testAddFirst() throws Exception + { + list.addFirst(a, 0); + list.addFirst(b, 0); + list.addFirst(c, 0); + list.addFirst(d, 0); + list.addFirst(e, 0); + + Assert.assertEquals(e, list.removeFirst()); + Assert.assertEquals(d, list.removeFirst()); + Assert.assertEquals(c, list.removeFirst()); + Assert.assertEquals(b, list.removeFirst()); + Assert.assertEquals(a, list.removeFirst()); + Assert.assertNull(list.removeFirst()); + } + + public void testAddLast() throws Exception + { + list.addLast(a, 0); + list.addLast(b, 0); + list.addLast(c, 0); + list.addLast(d, 0); + list.addLast(e, 0); + + Assert.assertEquals(a, list.removeFirst()); + Assert.assertEquals(b, list.removeFirst()); + Assert.assertEquals(c, list.removeFirst()); + Assert.assertEquals(d, list.removeFirst()); + Assert.assertEquals(e, list.removeFirst()); + Assert.assertNull(list.removeFirst()); + + } + + public void testAddLastAndFirst() throws Exception + { + list.addLast(a, 0); + list.addLast(b, 0); + list.addLast(c, 0); + list.addLast(d, 0); + list.addLast(e, 0); + list.addLast(f, 0); + list.addLast(g, 0); + list.addLast(h, 0); + list.addLast(i, 0); + list.addLast(j, 0); + + list.addFirst(k, 0); + list.addFirst(l, 0); + list.addFirst(m, 0); + list.addFirst(n, 0); + list.addFirst(o, 0); + list.addFirst(p, 0); + list.addFirst(q, 0); + list.addFirst(r, 0); + list.addFirst(s, 0); + list.addFirst(t, 0); + + assertEquals(t, list.removeFirst()); + assertEquals(s, list.removeFirst()); + assertEquals(r, list.removeFirst()); + assertEquals(q, list.removeFirst()); + assertEquals(p, list.removeFirst()); + assertEquals(o, list.removeFirst()); + assertEquals(n, list.removeFirst()); + assertEquals(m, list.removeFirst()); + assertEquals(l, list.removeFirst()); + assertEquals(k, list.removeFirst()); + + assertEquals(a, list.removeFirst()); + assertEquals(b, list.removeFirst()); + assertEquals(c, list.removeFirst()); + assertEquals(d, list.removeFirst()); + assertEquals(e, list.removeFirst()); + assertEquals(f, list.removeFirst()); + assertEquals(g, list.removeFirst()); + assertEquals(h, list.removeFirst()); + assertEquals(i, list.removeFirst()); + assertEquals(j, list.removeFirst()); + } + + public void testAddLastAndFirstWithIterator() throws Exception + { + list.addLast(a, 0); + list.addLast(b, 0); + list.addLast(c, 0); + list.addLast(d, 0); + list.addLast(e, 0); + list.addLast(f, 0); + list.addLast(g, 0); + list.addLast(h, 0); + list.addLast(i, 0); + list.addLast(j, 0); + + list.addFirst(k, 0); + list.addFirst(l, 0); + list.addFirst(m, 0); + list.addFirst(n, 0); + list.addFirst(o, 0); + list.addFirst(p, 0); + list.addFirst(q, 0); + list.addFirst(r, 0); + list.addFirst(s, 0); + list.addFirst(t, 0); + + HQIterator<Wibble> iter = list.iterator(); + + assertEquals(t, iter.next()); + assertEquals(s, iter.next()); + assertEquals(r, iter.next()); + assertEquals(q, iter.next()); + assertEquals(p, iter.next()); + assertEquals(o, iter.next()); + assertEquals(n, iter.next()); + assertEquals(m, iter.next()); + assertEquals(l, iter.next()); + assertEquals(k, iter.next()); + + + assertEquals(a, iter.next()); + assertEquals(b, iter.next()); + assertEquals(c, iter.next()); + assertEquals(d, iter.next()); + assertEquals(e, iter.next()); + assertEquals(f, iter.next()); + assertEquals(g, iter.next()); + assertEquals(h, iter.next()); + assertEquals(i, iter.next()); + assertEquals(j, iter.next()); + } + + public void testPeekFirst() + { + list.addLast(a, 0); + list.addLast(b, 1); + list.addLast(c, 2); + list.addLast(d, 3); + list.addLast(e, 4); + list.addLast(f, 5); + list.addLast(g, 6); + list.addLast(h, 7); + list.addLast(i, 8); + list.addLast(j, 9); + + Assert.assertEquals(j, list.peekFirst()); + Assert.assertEquals(j, list.peekFirst()); + + list.removeFirst(); + + Assert.assertEquals(i, list.peekFirst()); + Assert.assertEquals(i, list.peekFirst()); + + list.clear(); + } + + public void testRemoveFirst() throws Exception + { + list.addLast(a, 0); + list.addLast(b, 1); + list.addLast(c, 2); + list.addLast(d, 3); + list.addLast(e, 4); + list.addLast(f, 5); + list.addLast(g, 6); + list.addLast(h, 7); + list.addLast(i, 8); + list.addLast(j, 9); + + Assert.assertEquals(j, list.removeFirst()); + Assert.assertEquals(i, list.removeFirst()); + Assert.assertEquals(h, list.removeFirst()); + Assert.assertEquals(g, list.removeFirst()); + Assert.assertEquals(f, list.removeFirst()); + Assert.assertEquals(e, list.removeFirst()); + Assert.assertEquals(d, list.removeFirst()); + Assert.assertEquals(c, list.removeFirst()); + Assert.assertEquals(b, list.removeFirst()); + Assert.assertEquals(a, list.removeFirst()); + + Assert.assertNull(list.removeFirst()); + + list.addLast(a, 9); + list.addLast(b, 8); + list.addLast(c, 7); + list.addLast(d, 6); + list.addLast(e, 5); + list.addLast(f, 4); + list.addLast(g, 3); + list.addLast(h, 2); + list.addLast(i, 1); + list.addLast(j, 0); + + Assert.assertEquals(a, list.removeFirst()); + Assert.assertEquals(b, list.removeFirst()); + Assert.assertEquals(c, list.removeFirst()); + Assert.assertEquals(d, list.removeFirst()); + Assert.assertEquals(e, list.removeFirst()); + Assert.assertEquals(f, list.removeFirst()); + Assert.assertEquals(g, list.removeFirst()); + Assert.assertEquals(h, list.removeFirst()); + Assert.assertEquals(i, list.removeFirst()); + Assert.assertEquals(j, list.removeFirst()); + + Assert.assertNull(list.removeFirst()); + + list.addLast(a, 9); + list.addLast(b, 0); + list.addLast(c, 8); + list.addLast(d, 1); + list.addLast(e, 7); + list.addLast(f, 2); + list.addLast(g, 6); + list.addLast(h, 3); + list.addLast(i, 5); + list.addLast(j, 4); + + Assert.assertEquals(a, list.removeFirst()); + Assert.assertEquals(c, list.removeFirst()); + Assert.assertEquals(e, list.removeFirst()); + Assert.assertEquals(g, list.removeFirst()); + Assert.assertEquals(i, list.removeFirst()); + Assert.assertEquals(j, list.removeFirst()); + Assert.assertEquals(h, list.removeFirst()); + Assert.assertEquals(f, list.removeFirst()); + Assert.assertEquals(d, list.removeFirst()); + Assert.assertEquals(b, list.removeFirst()); + + Assert.assertNull(list.removeFirst()); + + list.addLast(a, 0); + list.addLast(b, 3); + list.addLast(c, 3); + list.addLast(d, 3); + list.addLast(e, 6); + list.addLast(f, 6); + list.addLast(g, 6); + list.addLast(h, 9); + list.addLast(i, 9); + list.addLast(j, 9); + + Assert.assertEquals(h, list.removeFirst()); + Assert.assertEquals(i, list.removeFirst()); + Assert.assertEquals(j, list.removeFirst()); + Assert.assertEquals(e, list.removeFirst()); + Assert.assertEquals(f, list.removeFirst()); + Assert.assertEquals(g, list.removeFirst()); + Assert.assertEquals(b, list.removeFirst()); + Assert.assertEquals(c, list.removeFirst()); + Assert.assertEquals(d, list.removeFirst()); + Assert.assertEquals(a, list.removeFirst()); + + Assert.assertNull(list.removeFirst()); + + list.addLast(a, 5); + list.addLast(b, 5); + list.addLast(c, 5); + list.addLast(d, 5); + list.addLast(e, 5); + list.addLast(f, 5); + list.addLast(g, 5); + list.addLast(h, 5); + list.addLast(i, 5); + list.addLast(j, 5); + + Assert.assertEquals(a, list.removeFirst()); + Assert.assertEquals(b, list.removeFirst()); + Assert.assertEquals(c, list.removeFirst()); + Assert.assertEquals(d, list.removeFirst()); + Assert.assertEquals(e, list.removeFirst()); + Assert.assertEquals(f, list.removeFirst()); + Assert.assertEquals(g, list.removeFirst()); + Assert.assertEquals(h, list.removeFirst()); + Assert.assertEquals(i, list.removeFirst()); + Assert.assertEquals(j, list.removeFirst()); + + Assert.assertNull(list.removeFirst()); + + list.addLast(j, 5); + list.addLast(i, 5); + list.addLast(h, 5); + list.addLast(g, 5); + list.addLast(f, 5); + list.addLast(e, 5); + list.addLast(d, 5); + list.addLast(c, 5); + list.addLast(b, 5); + list.addLast(a, 5); + + Assert.assertEquals(j, list.removeFirst()); + Assert.assertEquals(i, list.removeFirst()); + Assert.assertEquals(h, list.removeFirst()); + Assert.assertEquals(g, list.removeFirst()); + Assert.assertEquals(f, list.removeFirst()); + Assert.assertEquals(e, list.removeFirst()); + Assert.assertEquals(d, list.removeFirst()); + Assert.assertEquals(c, list.removeFirst()); + Assert.assertEquals(b, list.removeFirst()); + Assert.assertEquals(a, list.removeFirst()); + + Assert.assertNull(list.removeFirst()); + + } + + public void testIterator() + { + list.addLast(a, 9); + list.addLast(b, 9); + list.addLast(c, 8); + list.addLast(d, 8); + list.addLast(e, 7); + list.addLast(f, 7); + list.addLast(g, 7); + list.addLast(h, 6); + list.addLast(i, 6); + list.addLast(j, 6); + list.addLast(k, 5); + list.addLast(l, 5); + list.addLast(m, 4); + list.addLast(n, 4); + list.addLast(o, 4); + list.addLast(p, 3); + list.addLast(q, 3); + list.addLast(r, 3); + list.addLast(s, 2); + list.addLast(t, 2); + list.addLast(u, 2); + list.addLast(v, 1); + list.addLast(w, 1); + list.addLast(x, 1); + list.addLast(y, 0); + list.addLast(z, 0); + + HQIterator<Wibble> iter = list.iterator(); + + int c = 0; + Wibble w; + while ((w = iter.next()) != null) + { + c++; + } + Assert.assertEquals(c, 26); + Assert.assertEquals(26, list.size()); + + iter = list.iterator(); + + w = (Wibble)iter.next(); + Assert.assertEquals("a", w.s); + w = (Wibble)iter.next(); + Assert.assertEquals("b", w.s); + w = (Wibble)iter.next(); + Assert.assertEquals("c", w.s); + w = (Wibble)iter.next(); + Assert.assertEquals("d", w.s); + w = (Wibble)iter.next(); + Assert.assertEquals("e", w.s); + w = (Wibble)iter.next(); + Assert.assertEquals("f", w.s); + w = (Wibble)iter.next(); + Assert.assertEquals("g", w.s); + w = (Wibble)iter.next(); + Assert.assertEquals("h", w.s); + w = (Wibble)iter.next(); + Assert.assertEquals("i", w.s); + w = (Wibble)iter.next(); + Assert.assertEquals("j", w.s); + w = (Wibble)iter.next(); + Assert.assertEquals("k", w.s); + w = (Wibble)iter.next(); + Assert.assertEquals("l", w.s); + w = (Wibble)iter.next(); + Assert.assertEquals("m", w.s); + w = (Wibble)iter.next(); + Assert.assertEquals("n", w.s); + w = (Wibble)iter.next(); + Assert.assertEquals("o", w.s); + w = (Wibble)iter.next(); + Assert.assertEquals("p", w.s); + w = (Wibble)iter.next(); + Assert.assertEquals("q", w.s); + w = (Wibble)iter.next(); + Assert.assertEquals("r", w.s); + w = (Wibble)iter.next(); + Assert.assertEquals("s", w.s); + w = (Wibble)iter.next(); + Assert.assertEquals("t", w.s); + w = (Wibble)iter.next(); + Assert.assertEquals("u", w.s); + w = (Wibble)iter.next(); + Assert.assertEquals("v", w.s); + w = (Wibble)iter.next(); + Assert.assertEquals("w", w.s); + w = (Wibble)iter.next(); + Assert.assertEquals("x", w.s); + w = (Wibble)iter.next(); + Assert.assertEquals("y", w.s); + w = (Wibble)iter.next(); + Assert.assertEquals("z", w.s); + Assert.assertNull(iter.next()); + + iter = list.iterator(); + w = (Wibble)iter.next(); + Assert.assertEquals("a", w.s); + + iter.remove(); + + Assert.assertEquals(25, list.size()); + + w = (Wibble)iter.next(); + Assert.assertEquals("b", w.s); + w = (Wibble)iter.next(); + Assert.assertEquals("c", w.s); + w = (Wibble)iter.next(); + Assert.assertEquals("d", w.s); + + iter.remove(); + + Assert.assertEquals(24, list.size()); + + w = (Wibble)iter.next(); + Assert.assertEquals("e", w.s); + w = (Wibble)iter.next(); + Assert.assertEquals("f", w.s); + w = (Wibble)iter.next(); + Assert.assertEquals("g", w.s); + w = (Wibble)iter.next(); + Assert.assertEquals("h", w.s); + w = (Wibble)iter.next(); + Assert.assertEquals("i", w.s); + w = (Wibble)iter.next(); + Assert.assertEquals("j", w.s); + + iter.remove(); + + Assert.assertEquals(23, list.size()); + + w = (Wibble)iter.next(); + Assert.assertEquals("k", w.s); + w = (Wibble)iter.next(); + Assert.assertEquals("l", w.s); + w = (Wibble)iter.next(); + Assert.assertEquals("m", w.s); + w = (Wibble)iter.next(); + Assert.assertEquals("n", w.s); + w = (Wibble)iter.next(); + Assert.assertEquals("o", w.s); + w = (Wibble)iter.next(); + Assert.assertEquals("p", w.s); + w = (Wibble)iter.next(); + Assert.assertEquals("q", w.s); + w = (Wibble)iter.next(); + Assert.assertEquals("r", w.s); + w = (Wibble)iter.next(); + Assert.assertEquals("s", w.s); + w = (Wibble)iter.next(); + Assert.assertEquals("t", w.s); + w = (Wibble)iter.next(); + Assert.assertEquals("u", w.s); + w = (Wibble)iter.next(); + Assert.assertEquals("v", w.s); + w = (Wibble)iter.next(); + Assert.assertEquals("w", w.s); + w = (Wibble)iter.next(); + Assert.assertEquals("x", w.s); + w = (Wibble)iter.next(); + Assert.assertEquals("y", w.s); + w = (Wibble)iter.next(); + Assert.assertEquals("z", w.s); + iter.remove(); + Assert.assertNull(iter.next()); + + iter = list.iterator(); + w = (Wibble)iter.next(); + Assert.assertEquals("b", w.s); + w = (Wibble)iter.next(); + Assert.assertEquals("c", w.s); + w = (Wibble)iter.next(); + Assert.assertEquals("e", w.s); + w = (Wibble)iter.next(); + Assert.assertEquals("f", w.s); + w = (Wibble)iter.next(); + Assert.assertEquals("g", w.s); + w = (Wibble)iter.next(); + Assert.assertEquals("h", w.s); + w = (Wibble)iter.next(); + Assert.assertEquals("i", w.s); + w = (Wibble)iter.next(); + Assert.assertEquals("k", w.s); + w = (Wibble)iter.next(); + Assert.assertEquals("l", w.s); + w = (Wibble)iter.next(); + Assert.assertEquals("m", w.s); + w = (Wibble)iter.next(); + Assert.assertEquals("n", w.s); + w = (Wibble)iter.next(); + Assert.assertEquals("o", w.s); + w = (Wibble)iter.next(); + Assert.assertEquals("p", w.s); + w = (Wibble)iter.next(); + Assert.assertEquals("q", w.s); + w = (Wibble)iter.next(); + Assert.assertEquals("r", w.s); + w = (Wibble)iter.next(); + Assert.assertEquals("s", w.s); + w = (Wibble)iter.next(); + Assert.assertEquals("t", w.s); + w = (Wibble)iter.next(); + Assert.assertEquals("u", w.s); + w = (Wibble)iter.next(); + Assert.assertEquals("v", w.s); + w = (Wibble)iter.next(); + Assert.assertEquals("w", w.s); + w = (Wibble)iter.next(); + Assert.assertEquals("x", w.s); + w = (Wibble)iter.next(); + Assert.assertEquals("y", w.s); + Assert.assertNull(iter.next()); + + //We test again - should still be null (there was a bug here) + Assert.assertNull(iter.next()); + + } + + public void testClear() + { + list.addLast(a, 0); + list.addLast(b, 3); + list.addLast(c, 3); + list.addLast(d, 3); + list.addLast(e, 6); + list.addLast(f, 6); + list.addLast(g, 6); + list.addLast(h, 9); + list.addLast(i, 9); + list.addLast(j, 9); + + list.clear(); + + Assert.assertNull(list.removeFirst()); + } + + class Wibble + { + String s; + + Wibble(final String s) + { + this.s = s; + } + + public String toString() + { + return s; + } + } + +} Modified: trunk/tests/src/org/hornetq/tests/unit/core/postoffice/impl/BindingsImplTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/unit/core/postoffice/impl/BindingsImplTest.java 2010-03-22 23:43:26 UTC (rev 8946) +++ trunk/tests/src/org/hornetq/tests/unit/core/postoffice/impl/BindingsImplTest.java 2010-03-23 15:11:11 UTC (rev 8947) @@ -310,7 +310,7 @@ return 0; } - public int decrementRefCount(final MessageReference reference) throws Exception + public int decrementRefCount() throws Exception { // TODO Auto-generated method stub return 0; @@ -346,7 +346,7 @@ return 0; } - public int incrementRefCount(final MessageReference reference) throws Exception + public int incrementRefCount() throws Exception { // TODO Auto-generated method stub return 0; Modified: trunk/tests/src/org/hornetq/tests/unit/core/postoffice/impl/FakeQueue.java =================================================================== --- trunk/tests/src/org/hornetq/tests/unit/core/postoffice/impl/FakeQueue.java 2010-03-22 23:43:26 UTC (rev 8946) +++ trunk/tests/src/org/hornetq/tests/unit/core/postoffice/impl/FakeQueue.java 2010-03-23 15:11:11 UTC (rev 8947) @@ -273,7 +273,7 @@ /* (non-Javadoc) * @see org.hornetq.core.server.Queue#getMessagesAdded() */ - public int getMessagesAdded() + public long getMessagesAdded() { // TODO Auto-generated method stub return 0; @@ -425,10 +425,8 @@ /* (non-Javadoc) * @see org.hornetq.core.server.Queue#removeConsumer(org.hornetq.core.server.Consumer) */ - public boolean removeConsumer(final Consumer consumer) throws Exception + public void removeConsumer(final Consumer consumer) throws Exception { - // TODO Auto-generated method stub - return false; } /* (non-Javadoc) Deleted: trunk/tests/src/org/hornetq/tests/unit/core/remoting/impl/netty/ChannelBufferWrapperTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/unit/core/remoting/impl/netty/ChannelBufferWrapperTest.java 2010-03-22 23:43:26 UTC (rev 8946) +++ trunk/tests/src/org/hornetq/tests/unit/core/remoting/impl/netty/ChannelBufferWrapperTest.java 2010-03-23 15:11:11 UTC (rev 8947) @@ -1,58 +0,0 @@ -/* - * Copyright 2009 Red Hat, Inc. - * Red Hat licenses this file to you under the Apache License, version - * 2.0 (the "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * http://www.apache.org/licenses/LICENSE-2.0 - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or - * implied. See the License for the specific language governing - * permissions and limitations under the License. - */ - -package org.hornetq.tests.unit.core.remoting.impl.netty; - -import org.hornetq.api.core.HornetQBuffer; -import org.hornetq.core.buffers.impl.ChannelBufferWrapper; -import org.hornetq.tests.unit.core.remoting.HornetQBufferTestBase; -import org.jboss.netty.buffer.ChannelBuffer; -import org.jboss.netty.buffer.ChannelBuffers; - -/** - * @author <a href="mailto:jmesnil@redhat.com">Jeff Mesnil</a> - * - * @version <tt>$Revision$</tt> - * - */ -public class ChannelBufferWrapperTest extends HornetQBufferTestBase -{ - - // Constants ----------------------------------------------------- - - // Attributes ---------------------------------------------------- - - // Static -------------------------------------------------------- - - // Constructors -------------------------------------------------- - - // Public -------------------------------------------------------- - - // BufferWrapperBase overrides ----------------------------------- - - @Override - protected HornetQBuffer createBuffer() - { - ChannelBuffer buffer = ChannelBuffers.dynamicBuffer(512); - buffer.writerIndex(buffer.capacity()); - return new ChannelBufferWrapper(buffer); - } - - // Package protected --------------------------------------------- - - // Protected ----------------------------------------------------- - - // Private ------------------------------------------------------- - - // Inner classes ------------------------------------------------- -} Modified: trunk/tests/src/org/hornetq/tests/unit/core/remoting/impl/netty/NettyConnectionTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/unit/core/remoting/impl/netty/NettyConnectionTest.java 2010-03-22 23:43:26 UTC (rev 8946) +++ trunk/tests/src/org/hornetq/tests/unit/core/remoting/impl/netty/NettyConnectionTest.java 2010-03-23 15:11:11 UTC (rev 8947) @@ -47,7 +47,7 @@ public void testGetID() throws Exception { Channel channel = new SimpleChannel(RandomUtil.randomInt()); - NettyConnection conn = new NettyConnection(channel, new MyListener()); + NettyConnection conn = new NettyConnection(channel, new MyListener(), -1); Assert.assertEquals(channel.getId().intValue(), conn.getID()); } @@ -59,7 +59,7 @@ Assert.assertEquals(0, channel.getWritten().size()); - NettyConnection conn = new NettyConnection(channel, new MyListener()); + NettyConnection conn = new NettyConnection(channel, new MyListener(), -1); conn.write(buff); Assert.assertEquals(1, channel.getWritten().size()); @@ -68,7 +68,7 @@ public void testCreateBuffer() throws Exception { Channel channel = new SimpleChannel(RandomUtil.randomInt()); - NettyConnection conn = new NettyConnection(channel, new MyListener()); + NettyConnection conn = new NettyConnection(channel, new MyListener(), -1); final int size = 1234; Modified: trunk/tests/src/org/hornetq/tests/unit/core/server/impl/QueueImplTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/unit/core/server/impl/QueueImplTest.java 2010-03-22 23:43:26 UTC (rev 8946) +++ trunk/tests/src/org/hornetq/tests/unit/core/server/impl/QueueImplTest.java 2010-03-23 15:11:11 UTC (rev 8947) @@ -147,7 +147,7 @@ Assert.assertEquals(1, queue.getConsumerCount()); - Assert.assertTrue(queue.removeConsumer(cons1)); + queue.removeConsumer(cons1); Assert.assertEquals(0, queue.getConsumerCount()); @@ -159,23 +159,23 @@ Assert.assertEquals(3, queue.getConsumerCount()); - Assert.assertFalse(queue.removeConsumer(new FakeConsumer())); + queue.removeConsumer(new FakeConsumer()); Assert.assertEquals(3, queue.getConsumerCount()); - Assert.assertTrue(queue.removeConsumer(cons1)); + queue.removeConsumer(cons1); Assert.assertEquals(2, queue.getConsumerCount()); - Assert.assertTrue(queue.removeConsumer(cons2)); + queue.removeConsumer(cons2); Assert.assertEquals(1, queue.getConsumerCount()); - Assert.assertTrue(queue.removeConsumer(cons3)); + queue.removeConsumer(cons3); Assert.assertEquals(0, queue.getConsumerCount()); - Assert.assertFalse(queue.removeConsumer(cons3)); + queue.removeConsumer(cons3); } public void testGetFilter() @@ -252,44 +252,6 @@ } - public void testSimpleDirectDelivery() throws Exception - { - QueueImpl queue = new QueueImpl(1, - QueueImplTest.address1, - QueueImplTest.queue1, - null, - false, - true, - scheduledExecutor, - null, - null, - null, - executor); - - FakeConsumer consumer = new FakeConsumer(); - - queue.addConsumer(consumer); - - final int numMessages = 10; - - List<MessageReference> refs = new ArrayList<MessageReference>(); - - for (int i = 0; i < numMessages; i++) - { - MessageReference ref = generateReference(queue, i); - - refs.add(ref); - - queue.addLast(ref); - } - - Assert.assertEquals(numMessages, queue.getMessageCount()); - Assert.assertEquals(0, queue.getScheduledCount()); - Assert.assertEquals(numMessages, queue.getDeliveringCount()); - - assertRefListsIdenticalRefs(refs, consumer.getReferences()); - } - public void testSimpleNonDirectDelivery() throws Exception { QueueImpl queue = new QueueImpl(1, @@ -599,6 +561,8 @@ queue.addLast(ref); } + + queue.deliverNow(); Assert.assertEquals(numMessages * 2, queue.getMessageCount()); Assert.assertEquals(0, queue.getScheduledCount()); @@ -631,6 +595,8 @@ queue.addLast(ref); } + + queue.deliverNow(); Assert.assertEquals(numMessages * 3, queue.getMessageCount()); Assert.assertEquals(0, queue.getScheduledCount()); @@ -662,6 +628,8 @@ queue.addLast(ref); } + queue.deliverNow(); + Assert.assertEquals(numMessages * 2, queue.getMessageCount()); Assert.assertEquals(0, queue.getScheduledCount()); Assert.assertEquals(numMessages * 2, queue.getDeliveringCount()); @@ -688,6 +656,8 @@ queue.addLast(ref); } + + queue.deliverNow(); Assert.assertEquals(numMessages, queue.getMessageCount()); Assert.assertEquals(0, queue.getScheduledCount()); @@ -697,49 +667,7 @@ } - public void testConsumerReturningNull() throws Exception - { - QueueImpl queue = new QueueImpl(1, - QueueImplTest.address1, - QueueImplTest.queue1, - null, - false, - true, - scheduledExecutor, - null, - null, - null, - executor); - - class NullConsumer implements Consumer - { - public Filter getFilter() - { - return null; - } - - public HandleStatus handle(final MessageReference reference) - { - return null; - } - } - - queue.addConsumer(new NullConsumer()); - - MessageReference ref = generateReference(queue, 1); - - try - { - queue.addLast(ref); - - Assert.fail("Should throw IllegalStateException"); - } - catch (IllegalStateException e) - { - // Ok - } - } - + public void testRoundRobinWithQueueing() throws Exception { QueueImpl queue = new QueueImpl(1, @@ -793,57 +721,6 @@ } } - public void testRoundRobinDirect() throws Exception - { - QueueImpl queue = new QueueImpl(1, - QueueImplTest.address1, - QueueImplTest.queue1, - null, - false, - true, - scheduledExecutor, - null, - null, - null, - executor); - - final int numMessages = 10; - - List<MessageReference> refs = new ArrayList<MessageReference>(); - - FakeConsumer cons1 = new FakeConsumer(); - - FakeConsumer cons2 = new FakeConsumer(); - - queue.addConsumer(cons1); - - queue.addConsumer(cons2); - - queue.deliverNow(); - - for (int i = 0; i < numMessages; i++) - { - MessageReference ref = generateReference(queue, i); - - refs.add(ref); - - queue.addLast(ref); - } - - Assert.assertEquals(numMessages / 2, cons1.getReferences().size()); - - Assert.assertEquals(numMessages / 2, cons2.getReferences().size()); - - for (int i = 0; i < numMessages; i++) - { - MessageReference ref; - - ref = i % 2 == 0 ? cons1.getReferences().get(i / 2) : cons2.getReferences().get(i / 2); - - Assert.assertEquals(refs.get(i), ref); - } - } - public void testWithPriorities() throws Exception { QueueImpl queue = new QueueImpl(1, @@ -876,9 +753,9 @@ FakeConsumer consumer = new FakeConsumer(); queue.addConsumer(consumer); - + queue.deliverNow(); - + List<MessageReference> receivedRefs = consumer.getReferences(); // Should be in reverse order @@ -890,25 +767,6 @@ Assert.assertEquals(refs.get(i), receivedRefs.get(9 - i)); } - // But if we send more - since we are now in direct mode - the order will be the send order - // since the refs don't get queued - - consumer.clearReferences(); - - refs.clear(); - - for (int i = 0; i < numMessages; i++) - { - MessageReference ref = generateReference(queue, i); - - ref.getMessage().setPriority((byte)i); - - refs.add(ref); - - queue.addLast(ref); - } - - assertRefListsIdenticalRefs(refs, consumer.getReferences()); } public void testConsumerWithFiltersDirect() throws Exception Modified: trunk/tests/src/org/hornetq/tests/unit/core/server/impl/fakes/FakeQueueFactory.java =================================================================== --- trunk/tests/src/org/hornetq/tests/unit/core/server/impl/fakes/FakeQueueFactory.java 2010-03-22 23:43:26 UTC (rev 8946) +++ trunk/tests/src/org/hornetq/tests/unit/core/server/impl/fakes/FakeQueueFactory.java 2010-03-23 15:11:11 UTC (rev 8947) @@ -13,6 +13,7 @@ package org.hornetq.tests.unit.core.server.impl.fakes; +import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; @@ -33,6 +34,8 @@ public class FakeQueueFactory implements QueueFactory { private final ScheduledExecutorService scheduledExecutor = Executors.newSingleThreadScheduledExecutor(); + + private final ExecutorService executor = Executors.newSingleThreadExecutor(); private PostOffice postOffice; @@ -53,7 +56,7 @@ postOffice, null, null, - null); + executor); } public void setPostOffice(final PostOffice postOffice) @@ -65,6 +68,8 @@ public void stop() throws Exception { scheduledExecutor.shutdown(); + + executor.shutdown(); } } Added: trunk/tests/src/org/hornetq/tests/unit/util/concurrent/HornetQConcurrentLinkedQueueTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/unit/util/concurrent/HornetQConcurrentLinkedQueueTest.java (rev 0) +++ trunk/tests/src/org/hornetq/tests/unit/util/concurrent/HornetQConcurrentLinkedQueueTest.java 2010-03-23 15:11:11 UTC (rev 8947) @@ -0,0 +1,286 @@ +/* + * Copyright 2010 Red Hat, Inc. + * Red Hat licenses this file to you under the Apache License, version + * 2.0 (the "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * http://www.apache.org/licenses/LICENSE-2.0 + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or + * implied. See the License for the specific language governing + * permissions and limitations under the License. + */ + +package org.hornetq.tests.unit.util.concurrent; + +import java.util.Iterator; +import java.util.Queue; + +import junit.framework.TestCase; + +import org.hornetq.core.logging.Logger; +import org.hornetq.utils.concurrent.HQIterator; +import org.hornetq.utils.concurrent.HornetQConcurrentLinkedQueue; + +/** + * A HornetQConcurrentLinkedQueueTest + * + * @author Tim Fox + * + * + */ +public class HornetQConcurrentLinkedQueueTest extends TestCase +{ + private static final Logger log = Logger.getLogger(HornetQConcurrentLinkedQueueTest.class); + + public void testIteratorSeesNewNodes() + { + Queue<Object> queue = new HornetQConcurrentLinkedQueue<Object>(); + + for (int i = 0; i < 10; i++) + { + queue.add(i); + } + + Iterator<Object> iter = queue.iterator(); + + for (int i = 0; i < 10; i++) + { + assertTrue(iter.hasNext()); + assertEquals(i, iter.next()); + } + + assertFalse(iter.hasNext()); + assertFalse(iter.hasNext()); + + //Make sure iterator picks up any new nodes added later, without having to reset it + + for (int i = 10; i < 20; i++) + { + queue.add(i); + assertTrue(iter.hasNext()); + } + + for (int i = 10; i < 20; i++) + { + assertTrue(iter.hasNext()); + assertEquals(i, iter.next()); + } + + assertFalse(iter.hasNext()); + assertFalse(iter.hasNext()); + } + + public void testDeletedNodesNotSeenByIterator() + { + Queue<Object> queue = new HornetQConcurrentLinkedQueue<Object>(); + + for (int i = 0; i < 10; i++) + { + queue.add(i); + } + + Iterator<Object> iter = queue.iterator(); + + for (int i = 0; i < 10; i++) + { + assertTrue(iter.hasNext()); + assertEquals(i, iter.next()); + } + + assertFalse(iter.hasNext()); + assertFalse(iter.hasNext()); + + //Now add some more nodes + + for (int i = 10; i < 20; i++) + { + queue.add(i); + } + + //Create another iterator + + Iterator<Object> iter2 = queue.iterator(); + + //Iterate past the first 10 + + for (int i = 0; i < 11; i++) + { + assertTrue(iter2.hasNext()); + assertEquals(i, iter2.next()); + } + + //Remove the 10th element + + iter2.remove(); + + //Now make sure the first iterator can skip past this + + for (int i = 11; i < 20; i++) + { + assertTrue(iter.hasNext()); + assertEquals(i, iter.next()); + } + + assertFalse(iter.hasNext()); + + //Add further nodes + + for (int i = 20; i < 30; i++) + { + queue.add(i); + } + + assertTrue(iter.hasNext()); + + //Now delete them all with iter2 + + for (int i = 11; i < 30; i++) + { + assertTrue(iter2.hasNext()); + assertEquals(i, iter2.next()); + iter2.remove(); + } + + //Since it returned true before it must return true again - this is part of the contract of Iterator + assertTrue(iter.hasNext()); + + assertEquals(20, iter.next()); // Even though it was deleted + + assertFalse(iter.hasNext()); + + } + + public void testIteratorSeesNewNodesHQIterator() + { + HornetQConcurrentLinkedQueue<Object> queue = new HornetQConcurrentLinkedQueue<Object>(); + + for (int i = 0; i < 10; i++) + { + queue.add(i); + } + + HQIterator<Object> iter = queue.hqIterator(); + + for (int i = 0; i < 10; i++) + { + assertEquals(i, iter.next()); + } + + assertNull(iter.next()); + assertNull(iter.next()); + + //Make sure iterator picks up any new nodes added later, without having to reset it + + for (int i = 10; i < 20; i++) + { + queue.add(i); + } + + for (int i = 10; i < 20; i++) + { + assertEquals(i, iter.next()); + } + + assertNull(iter.next()); + assertNull(iter.next()); + } + + public void testDeletedNodesNotSeenByIteratorHQIterator() + { + HornetQConcurrentLinkedQueue<Object> queue = new HornetQConcurrentLinkedQueue<Object>(); + + HQIterator<Object> iter = queue.hqIterator(); + + assertNull(iter.next()); + + for (int i = 0; i < 10; i++) + { + queue.add(i); + } + + for (int i = 0; i < 10; i++) + { + assertEquals(i, iter.next()); + } + + assertNull(iter.next()); + + assertNull(iter.next()); + + //Now add some more nodes + + for (int i = 10; i < 20; i++) + { + queue.add(i); + } + + //Create another iterator + + HQIterator<Object> iter2 = queue.hqIterator(); + + //Iterate past the first 10 + + for (int i = 0; i < 11; i++) + { + assertEquals(i, iter2.next()); + } + + //Remove the 10th element + + iter2.remove(); + + //Now make sure the first iterator can skip past this + + for (int i = 11; i < 20; i++) + { + assertEquals(i, iter.next()); + } + + assertNull(iter.next()); + + //Add further nodes + + for (int i = 20; i < 30; i++) + { + queue.add(i); + } + + assertEquals(20, iter.next()); + + //Now delete them all bar one with iter2 + + for (int i = 11; i < 29; i++) + { + assertEquals(i, iter2.next()); + iter2.remove(); + } + + assertEquals(29, iter.next()); + + iter.remove(); + + assertNull(iter2.next()); + } + + + public void testCallingHasNextTwiceDoesntResetIterator() + { + HornetQConcurrentLinkedQueue<Object> queue = new HornetQConcurrentLinkedQueue<Object>(); + + for (int i = 0; i < 10; i++) + { + queue.add(i); + } + + HQIterator<Object> iter = queue.hqIterator(); + + for (int i = 0; i < 10; i++) + { + assertEquals(i, iter.next()); + } + + assertNull(iter.next()); + assertNull(iter.next()); + } +} Modified: trunk/tests/src/org/hornetq/tests/util/JMSTestBase.java =================================================================== --- trunk/tests/src/org/hornetq/tests/util/JMSTestBase.java 2010-03-22 23:43:26 UTC (rev 8946) +++ trunk/tests/src/org/hornetq/tests/util/JMSTestBase.java 2010-03-23 15:11:11 UTC (rev 8947) @@ -157,7 +157,7 @@ // Inner classes ------------------------------------------------- - private void registerConnectionFactory() throws Exception + protected void registerConnectionFactory() throws Exception { List<Pair<TransportConfiguration, TransportConfiguration>> connectorConfigs = new ArrayList<Pair<TransportConfiguration, TransportConfiguration>>(); connectorConfigs.add(new Pair<TransportConfiguration, TransportConfiguration>(new TransportConfiguration(NettyConnectorFactory.class.getName()),