[hornetq-commits] JBoss hornetq SVN: r8171 - in trunk: examples/core/perf and 58 other directories.

Author: timfox Date: 2009-10-30 13:41:19 -0400 (Fri, 30 Oct 2009) New Revision: 8171 Added: trunk/src/main/org/hornetq/core/client/SessionFailureListener.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/remoting/impl/wireformat/SessionProducerCreditsMessage.java trunk/src/main/org/hornetq/core/remoting/impl/wireformat/SessionRequestProducerCreditsMessage.java trunk/src/main/org/hornetq/core/server/impl/ServerProducerCreditManager.java trunk/src/main/org/hornetq/core/server/impl/ServerProducerCreditManagerImpl.java trunk/tests/src/org/hornetq/tests/integration/EncodeSizeTest.java trunk/tests/src/org/hornetq/tests/integration/client/ProducerFlowControlTest.java trunk/tests/src/org/hornetq/tests/integration/client/TransactionalSendTest.java Removed: trunk/tests/src/org/hornetq/tests/integration/client/SendTest.java Modified: trunk/docs/user-manual/en/appserver-integration.xml trunk/docs/user-manual/en/command-buffering.xml trunk/docs/user-manual/en/configuration-index.xml trunk/examples/core/perf/perf.properties trunk/examples/core/perf/server0/hornetq-configuration.xml trunk/examples/core/perf/src/org/hornetq/core/example/PerfBase.java trunk/examples/core/perf/src/org/hornetq/core/example/PerfParams.java trunk/src/config/common/schema/hornetq-configuration.xsd trunk/src/config/common/schema/hornetq-jms.xsd trunk/src/config/ra.xml trunk/src/main/org/hornetq/core/client/ClientSession.java trunk/src/main/org/hornetq/core/client/ClientSessionFactory.java trunk/src/main/org/hornetq/core/client/impl/ClientConsumerImpl.java trunk/src/main/org/hornetq/core/client/impl/ClientMessageImpl.java trunk/src/main/org/hornetq/core/client/impl/ClientProducerImpl.java trunk/src/main/org/hornetq/core/client/impl/ClientSessionFactoryImpl.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/FailoverManager.java trunk/src/main/org/hornetq/core/client/impl/FailoverManagerImpl.java trunk/src/main/org/hornetq/core/config/impl/Validators.java trunk/src/main/org/hornetq/core/deployers/impl/AddressSettingsDeployer.java trunk/src/main/org/hornetq/core/journal/impl/NIOSequentialFile.java trunk/src/main/org/hornetq/core/journal/impl/SyncSpeedTest.java trunk/src/main/org/hornetq/core/message/Message.java trunk/src/main/org/hornetq/core/message/impl/MessageImpl.java trunk/src/main/org/hornetq/core/paging/PagingStore.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/JournalLargeServerMessage.java trunk/src/main/org/hornetq/core/persistence/impl/journal/JournalStorageManager.java trunk/src/main/org/hornetq/core/persistence/impl/nullpm/NullStorageLargeServerMessage.java trunk/src/main/org/hornetq/core/persistence/impl/nullpm/NullStorageManager.java trunk/src/main/org/hornetq/core/postoffice/Bindings.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/remoting/Channel.java trunk/src/main/org/hornetq/core/remoting/FailureListener.java trunk/src/main/org/hornetq/core/remoting/RemotingConnection.java trunk/src/main/org/hornetq/core/remoting/impl/ChannelImpl.java trunk/src/main/org/hornetq/core/remoting/impl/PacketDecoder.java trunk/src/main/org/hornetq/core/remoting/impl/RemotingConnectionImpl.java trunk/src/main/org/hornetq/core/remoting/impl/wireformat/PacketImpl.java trunk/src/main/org/hornetq/core/remoting/server/impl/RemotingServiceImpl.java trunk/src/main/org/hornetq/core/replication/impl/ReplicationManagerImpl.java trunk/src/main/org/hornetq/core/server/HornetQServer.java trunk/src/main/org/hornetq/core/server/LargeServerMessage.java trunk/src/main/org/hornetq/core/server/RoutingContext.java trunk/src/main/org/hornetq/core/server/ServerMessage.java trunk/src/main/org/hornetq/core/server/ServerSession.java trunk/src/main/org/hornetq/core/server/cluster/impl/BridgeImpl.java trunk/src/main/org/hornetq/core/server/cluster/impl/Redistributor.java trunk/src/main/org/hornetq/core/server/impl/DivertImpl.java trunk/src/main/org/hornetq/core/server/impl/HornetQPacketHandler.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/ServerMessageImpl.java trunk/src/main/org/hornetq/core/server/impl/ServerSessionImpl.java trunk/src/main/org/hornetq/core/server/impl/ServerSessionPacketHandler.java trunk/src/main/org/hornetq/core/settings/impl/AddressSettings.java trunk/src/main/org/hornetq/jms/client/HornetQConnection.java trunk/src/main/org/hornetq/jms/client/HornetQConnectionFactory.java trunk/src/main/org/hornetq/jms/server/JMSServerManager.java trunk/src/main/org/hornetq/jms/server/config/ConnectionFactoryConfiguration.java trunk/src/main/org/hornetq/jms/server/config/impl/ConnectionFactoryConfigurationImpl.java trunk/src/main/org/hornetq/jms/server/impl/JMSServerDeployer.java trunk/src/main/org/hornetq/jms/server/impl/JMSServerManagerImpl.java trunk/src/main/org/hornetq/jms/server/management/ConnectionFactoryControl.java trunk/src/main/org/hornetq/jms/server/management/JMSServerControl.java trunk/src/main/org/hornetq/jms/server/management/impl/JMSConnectionFactoryControlImpl.java trunk/src/main/org/hornetq/jms/server/management/impl/JMSServerControlImpl.java trunk/src/main/org/hornetq/jms/server/recovery/HornetQXAResourceWrapper.java trunk/src/main/org/hornetq/ra/ConnectionFactoryProperties.java trunk/src/main/org/hornetq/ra/HornetQRAManagedConnectionFactory.java trunk/src/main/org/hornetq/ra/HornetQResourceAdapter.java trunk/tests/config/hornetq-jms-for-JMSServerDeployerTest.xml trunk/tests/jms-tests/src/org/hornetq/jms/tests/BrowserTest.java trunk/tests/jms-tests/src/org/hornetq/jms/tests/CTSMiscellaneousTest.java trunk/tests/jms-tests/src/org/hornetq/jms/tests/JMSTestCase.java trunk/tests/jms-tests/src/org/hornetq/jms/tests/message/MessageHeaderTest.java trunk/tests/jms-tests/src/org/hornetq/jms/tests/tools/container/LocalTestServer.java trunk/tests/src/org/hornetq/tests/integration/client/AddressSettingsTest.java trunk/tests/src/org/hornetq/tests/integration/client/ConsumerTest.java trunk/tests/src/org/hornetq/tests/integration/client/HornetQCrashTest.java trunk/tests/src/org/hornetq/tests/integration/client/LargeMessageTest.java trunk/tests/src/org/hornetq/tests/integration/client/PagingTest.java trunk/tests/src/org/hornetq/tests/integration/client/ProducerTest.java trunk/tests/src/org/hornetq/tests/integration/client/SessionFactoryTest.java trunk/tests/src/org/hornetq/tests/integration/client/SessionSendAcknowledgementHandlerTest.java trunk/tests/src/org/hornetq/tests/integration/client/SessionTest.java trunk/tests/src/org/hornetq/tests/integration/cluster/bridge/BridgeReconnectTest.java trunk/tests/src/org/hornetq/tests/integration/cluster/bridge/BridgeStartTest.java trunk/tests/src/org/hornetq/tests/integration/cluster/distribution/ClusteredGroupingTest.java trunk/tests/src/org/hornetq/tests/integration/cluster/failover/AsynchronousFailoverTest.java trunk/tests/src/org/hornetq/tests/integration/cluster/failover/FailoverTest.java trunk/tests/src/org/hornetq/tests/integration/cluster/failover/GroupingFailoverSharedServerTest.java trunk/tests/src/org/hornetq/tests/integration/cluster/failover/PagingFailoverTest.java trunk/tests/src/org/hornetq/tests/integration/cluster/reattach/MultiThreadRandomReattachTestBase.java trunk/tests/src/org/hornetq/tests/integration/cluster/reattach/NettyMultiThreadRandomReattachTest.java trunk/tests/src/org/hornetq/tests/integration/cluster/reattach/RandomReattachTest.java trunk/tests/src/org/hornetq/tests/integration/cluster/reattach/ReattachTest.java trunk/tests/src/org/hornetq/tests/integration/jms/FloodServerTest.java trunk/tests/src/org/hornetq/tests/integration/jms/HornetQConnectionFactoryTest.java trunk/tests/src/org/hornetq/tests/integration/jms/ManualReconnectionToSingleServerTest.java trunk/tests/src/org/hornetq/tests/integration/jms/client/PreACKJMSTest.java trunk/tests/src/org/hornetq/tests/integration/jms/client/ResendTest.java trunk/tests/src/org/hornetq/tests/integration/jms/divert/DivertAndACKClientTest.java trunk/tests/src/org/hornetq/tests/integration/jms/server/JMSServerDeployerTest.java trunk/tests/src/org/hornetq/tests/integration/jms/server/management/JMSServerControlTest.java trunk/tests/src/org/hornetq/tests/integration/jms/server/management/JMSServerControlUsingJMSTest.java trunk/tests/src/org/hornetq/tests/integration/largemessage/LargeMessageTestBase.java trunk/tests/src/org/hornetq/tests/integration/remoting/PingTest.java trunk/tests/src/org/hornetq/tests/integration/replication/ReplicationTest.java trunk/tests/src/org/hornetq/tests/unit/core/postoffice/impl/BindingsImplTest.java trunk/tests/src/org/hornetq/tests/unit/core/server/impl/fakes/FakePostOffice.java trunk/tests/src/org/hornetq/tests/unit/core/settings/impl/AddressSettingsTest.java trunk/tests/src/org/hornetq/tests/unit/ra/ResourceAdapterTest.java trunk/tests/src/org/hornetq/tests/util/JMSTestBase.java Log: mainly producer flow control Modified: trunk/docs/user-manual/en/appserver-integration.xml =================================================================== --- trunk/docs/user-manual/en/appserver-integration.xml 2009-10-30 15:12:21 UTC (rev 8170) +++ trunk/docs/user-manual/en/appserver-integration.xml 2009-10-30 17:41:19 UTC (rev 8171) @@ -507,9 +507,9 @@ <entry>The max rate a consumer can receive.</entry> </row> <row> - <entry>ProducerWindowSize</entry> + <entry>ConfirmationWindowSize</entry> <entry>integer</entry> - <entry>The window size for the sending of messages.</entry> + <entry>The window size for confirmations.</entry> </row> <row> <entry>ProducerMaxRate</entry> Modified: trunk/docs/user-manual/en/command-buffering.xml =================================================================== --- trunk/docs/user-manual/en/command-buffering.xml 2009-10-30 15:12:21 UTC (rev 8170) +++ trunk/docs/user-manual/en/command-buffering.xml 2009-10-30 17:41:19 UTC (rev 8171) @@ -32,7 +32,7 @@ client can then free up space in the buffer.</para> <para>If you are using JMS and you're using the JMS service on the server to load your JMS connection factory instances into JNDI then this parameter can be configured in <literal - >hornetq-jms.xml</literal> using the element <literal>producer-window-size</literal> a. + >hornetq-jms.xml</literal> using the element <literal>confirmation-window-size</literal> a. If you're using JMS but not using JNDI then you can set these values directly on the <literal>HornetQConnectionFactory</literal> instance using the appropriate setter method.</para> Modified: trunk/docs/user-manual/en/configuration-index.xml =================================================================== --- trunk/docs/user-manual/en/configuration-index.xml 2009-10-30 15:12:21 UTC (rev 8170) +++ trunk/docs/user-manual/en/configuration-index.xml 2009-10-30 17:41:19 UTC (rev 8171) @@ -986,7 +986,7 @@ </row> <row> <entry><link linkend="command-buffering" - >connection-factory.producer-window-size</link></entry> + >connection-factory.confirmation-window-size</link></entry> <entry>Integer</entry> <entry>the window size (in bytes) for sending messages</entry> <entry>1024 * 1024</entry> Modified: trunk/examples/core/perf/perf.properties =================================================================== --- trunk/examples/core/perf/perf.properties 2009-10-30 15:12:21 UTC (rev 8170) +++ trunk/examples/core/perf/perf.properties 2009-10-30 17:41:19 UTC (rev 8171) @@ -1,18 +1,20 @@ -num-messages=10000 -num-warmup-messages=100 +num-messages=1000000 +num-warmup-messages=0 message-size=1024 -durable=true +durable=false transacted=false -batch-size=1000 -drain-queue=true +batch-size=1048576 +drain-queue=false throttle-rate=-1 address=perfAddress queue-name=perfQueue host=localhost port=5445 -tcp-buffer=1048576 -tcp-no-delay=true -send-window=1048576 -pre-ack=true +tcp-buffer=2048576 +tcp-no-delay=false +confirmation-window=-1 +producer-window=1048576 +consumer-window=1048576 +pre-ack=false block-ack=false -block-persistent=true +block-persistent=false Modified: trunk/examples/core/perf/server0/hornetq-configuration.xml =================================================================== --- trunk/examples/core/perf/server0/hornetq-configuration.xml 2009-10-30 15:12:21 UTC (rev 8170) +++ trunk/examples/core/perf/server0/hornetq-configuration.xml 2009-10-30 17:41:19 UTC (rev 8171) @@ -15,15 +15,15 @@ <security-enabled>false</security-enabled> - <persistence-enabled>true</persistence-enabled> + <persistence-enabled>false</persistence-enabled> - <journal-sync-non-transactional>true</journal-sync-non-transactional> - <journal-sync-transactional>true</journal-sync-transactional> + <journal-sync-non-transactional>false</journal-sync-non-transactional> + <journal-sync-transactional>false</journal-sync-transactional> <journal-type>ASYNCIO</journal-type> <journal-min-files>20</journal-min-files> <journal-aio-buffer-timeout>20000</journal-aio-buffer-timeout> - <log-journal-write-rate>true</log-journal-write-rate> - <run-sync-speed-test>true</run-sync-speed-test> + <log-journal-write-rate>false</log-journal-write-rate> + <run-sync-speed-test>false</run-sync-speed-test> <!-- <perf-blast-pages>5000</perf-blast-pages> --> @@ -33,4 +33,11 @@ </queue> </queues> + <address-settings> + <address-setting match="perfAddress"> + <max-size-bytes>10485760</max-size-bytes> + <address-full-policy>BLOCK</address-full-policy> + </address-setting> + </address-settings> + </configuration> Modified: trunk/examples/core/perf/src/org/hornetq/core/example/PerfBase.java =================================================================== --- trunk/examples/core/perf/src/org/hornetq/core/example/PerfBase.java 2009-10-30 15:12:21 UTC (rev 8170) +++ trunk/examples/core/perf/src/org/hornetq/core/example/PerfBase.java 2009-10-30 17:41:19 UTC (rev 8171) @@ -118,7 +118,9 @@ int tcpBufferSize = Integer.valueOf(props.getProperty("tcp-buffer")); boolean tcpNoDelay = Boolean.valueOf(props.getProperty("tcp-no-delay")); boolean preAck = Boolean.valueOf(props.getProperty("pre-ack")); - int sendWindowSize = Integer.valueOf(props.getProperty("send-window")); + int confirmationWindowSize = Integer.valueOf(props.getProperty("confirmation-window")); + int producerWindowSize = Integer.valueOf(props.getProperty("producer-window")); + int consumerWindowSize = Integer.valueOf(props.getProperty("consumer-window")); boolean blockOnACK = Boolean.valueOf(props.getProperty("block-ack", "false")); boolean blockOnPersistent = Boolean.valueOf(props.getProperty("block-persistent", "false")); @@ -137,7 +139,9 @@ log.info("tcp buffer: " + tcpBufferSize); log.info("tcp no delay: " + tcpNoDelay); log.info("pre-ack: " + preAck); - log.info("send-window: " + sendWindowSize); + log.info("confirmation-window: " + confirmationWindowSize); + log.info("producer-window: " + producerWindowSize); + log.info("consumer-window: " + consumerWindowSize); log.info("block-ack:" + blockOnACK); log.info("block-persistent:" + blockOnPersistent); @@ -157,7 +161,9 @@ perfParams.setTcpBufferSize(tcpBufferSize); perfParams.setTcpNoDelay(tcpNoDelay); perfParams.setPreAck(preAck); - perfParams.setSendWindow(sendWindowSize); + perfParams.setConfirmationWindow(confirmationWindowSize); + perfParams.setProducerWindow(producerWindowSize); + perfParams.setConsumerWindow(consumerWindowSize); perfParams.setBlockOnACK(blockOnACK); perfParams.setBlockOnPersistent(blockOnPersistent); @@ -190,7 +196,9 @@ factory = new ClientSessionFactoryImpl(new TransportConfiguration(NettyConnectorFactory.class.getName(), params)); factory.setPreAcknowledge(perfParams.isPreAck()); - factory.setProducerWindowSize(perfParams.getSendWindow()); + factory.setConfirmationWindowSize(perfParams.getConfirmationWindow()); + factory.setProducerWindowSize(perfParams.getProducerWindow()); + factory.setConsumerWindowSize(perfParams.getConsumerWindow()); factory.setAckBatchSize(perfParams.getBatchSize()); Modified: trunk/examples/core/perf/src/org/hornetq/core/example/PerfParams.java =================================================================== --- trunk/examples/core/perf/src/org/hornetq/core/example/PerfParams.java 2009-10-30 15:12:21 UTC (rev 8170) +++ trunk/examples/core/perf/src/org/hornetq/core/example/PerfParams.java 2009-10-30 17:41:19 UTC (rev 8171) @@ -39,7 +39,9 @@ private int tcpBufferSize; private boolean tcpNoDelay; private boolean preAck; - private int sendWindow; + private int confirmationWindow; + private int producerWindow; + private int consumerWindow; private boolean blockOnPersistent = true; private boolean blockOnACK = true; @@ -222,15 +224,33 @@ this.preAck = preAck; } - public synchronized int getSendWindow() + public synchronized int getConfirmationWindow() { - return sendWindow; + return confirmationWindow; } - public synchronized void setSendWindow(int sendWindow) + public synchronized void setConfirmationWindow(int confirmationWindow) { - this.sendWindow = sendWindow; + this.confirmationWindow = confirmationWindow; } + public int getProducerWindow() + { + return producerWindow; + } + public void setProducerWindow(int producerWindow) + { + this.producerWindow = producerWindow; + } + + public int getConsumerWindow() + { + return consumerWindow; + } + + public void setConsumerWindow(int consumerWindow) + { + this.consumerWindow = consumerWindow; + } } Modified: trunk/src/config/common/schema/hornetq-configuration.xsd =================================================================== --- trunk/src/config/common/schema/hornetq-configuration.xsd 2009-10-30 15:12:21 UTC (rev 8170) +++ trunk/src/config/common/schema/hornetq-configuration.xsd 2009-10-30 17:41:19 UTC (rev 8171) @@ -460,8 +460,8 @@ </xsd:element> <xsd:element maxOccurs="1" minOccurs="0" name="page-size-bytes" type="xsd:int"> </xsd:element> - <xsd:element maxOccurs="1" minOccurs="0" name="drop-messages-when-full" type="xsd:boolean"> - </xsd:element> + <xsd:element maxOccurs="1" minOccurs="0" name="address-full-policy" type="addressFullMessagePolicyType"> + </xsd:element> <xsd:element maxOccurs="1" minOccurs="0" name="distribution-policy-class" type="xsd:string"> </xsd:element> <xsd:element maxOccurs="1" minOccurs="0" name="message-counter-history-day-limit" type="xsd:int"> @@ -507,4 +507,14 @@ <xsd:attribute name="connector-name" type="xsd:IDREF" use="required"> </xsd:attribute> </xsd:complexType> + + <xsd:simpleType name="addressFullMessagePolicyType"> + <xsd:restriction base="xsd:string"> + <xsd:enumeration value="DROP"/> + <xsd:enumeration value="PAGE"/> + <xsd:enumeration value="BLOCK"/> + </xsd:restriction> + </xsd:simpleType> + + </xsd:schema> Modified: trunk/src/config/common/schema/hornetq-jms.xsd =================================================================== --- trunk/src/config/common/schema/hornetq-jms.xsd 2009-10-30 15:12:21 UTC (rev 8170) +++ trunk/src/config/common/schema/hornetq-jms.xsd 2009-10-30 17:41:19 UTC (rev 8171) @@ -54,7 +54,7 @@ <xsd:element name="consumer-max-rate" type="xsd:int" maxOccurs="1" minOccurs="0"> </xsd:element> - <xsd:element name="producer-window-size" type="xsd:int" + <xsd:element name="confirmation-window-size" type="xsd:int" maxOccurs="1" minOccurs="0"> </xsd:element> <xsd:element name="producer-max-rate" type="xsd:int" Modified: trunk/src/config/ra.xml =================================================================== --- trunk/src/config/ra.xml 2009-10-30 15:12:21 UTC (rev 8170) +++ trunk/src/config/ra.xml 2009-10-30 17:41:19 UTC (rev 8171) @@ -161,8 +161,8 @@ <config-property-value></config-property-value> </config-property> <config-property> - <description>The producer window size</description> - <config-property-name>ProducerWindowSize</config-property-name> + <description>The confirmation window size</description> + <config-property-name>ConfirmationWindowSize</config-property-name> <config-property-type>java.lang.Integer</config-property-type> <config-property-value></config-property-value> </config-property> Modified: trunk/src/main/org/hornetq/core/client/ClientSession.java =================================================================== --- trunk/src/main/org/hornetq/core/client/ClientSession.java 2009-10-30 15:12:21 UTC (rev 8170) +++ trunk/src/main/org/hornetq/core/client/ClientSession.java 2009-10-30 17:41:19 UTC (rev 8171) @@ -19,6 +19,7 @@ import org.hornetq.core.remoting.FailureListener; import org.hornetq.core.remoting.impl.wireformat.SessionBindingQueryResponseMessage; import org.hornetq.core.remoting.impl.wireformat.SessionQueueQueryResponseMessage; +import org.hornetq.core.remoting.spi.HornetQBuffer; import org.hornetq.utils.SimpleString; /* @@ -42,7 +43,7 @@ * Queues created by this method are <em>not</em> temporary */ void createQueue(String address, String queueName) throws HornetQException; - + void createQueue(String address, String queueName, boolean durable) throws HornetQException; void createQueue(SimpleString address, SimpleString queueName, SimpleString filterString, boolean durable) throws HornetQException; @@ -66,7 +67,7 @@ ClientConsumer createConsumer(SimpleString queueName, SimpleString filterString) throws HornetQException; ClientConsumer createConsumer(SimpleString queueName, SimpleString filterString, boolean browseOnly) throws HornetQException; - + ClientConsumer createConsumer(SimpleString queueName, boolean browseOnly) throws HornetQException; ClientConsumer createConsumer(SimpleString queueName, @@ -80,7 +81,7 @@ ClientConsumer createConsumer(String queueName, String filterString) throws HornetQException; ClientConsumer createConsumer(String queueName, String filterString, boolean browseOnly) throws HornetQException; - + ClientConsumer createConsumer(String queueName, boolean browseOnly) throws HornetQException; ClientConsumer createConsumer(String queueName, String filterString, int windowSize, int maxRate, boolean browseOnly) throws HornetQException; @@ -141,23 +142,25 @@ boolean isXA(); - ClientMessage createClientMessage(final byte type, - final boolean durable, - final long expiration, - final long timestamp, - final byte priority); + ClientMessage createClientMessage(byte type, boolean durable, long expiration, long timestamp, byte priority); - ClientMessage createClientMessage(final byte type, final boolean durable); + ClientMessage createClientMessage(byte type, boolean durable); - ClientMessage createClientMessage(final boolean durable); + ClientMessage createClientMessage(boolean durable); + ClientMessage createClientMessage(boolean durable, HornetQBuffer buffer); + + HornetQBuffer createBuffer(byte[] bytes); + + HornetQBuffer createBuffer(int size); + void start() throws HornetQException; void stop() throws HornetQException; - void addFailureListener(FailureListener listener); + void addFailureListener(SessionFailureListener listener); - boolean removeFailureListener(FailureListener listener); + boolean removeFailureListener(SessionFailureListener listener); int getVersion(); Modified: trunk/src/main/org/hornetq/core/client/ClientSessionFactory.java =================================================================== --- trunk/src/main/org/hornetq/core/client/ClientSessionFactory.java 2009-10-30 15:12:21 UTC (rev 8170) +++ trunk/src/main/org/hornetq/core/client/ClientSessionFactory.java 2009-10-30 17:41:19 UTC (rev 8171) @@ -82,10 +82,14 @@ void setConsumerMaxRate(int consumerMaxRate); + int getConfirmationWindowSize(); + + void setConfirmationWindowSize(int confirmationWindowSize); + int getProducerWindowSize(); void setProducerWindowSize(int producerWindowSize); - + int getProducerMaxRate(); void setProducerMaxRate(int producerMaxRate); Added: trunk/src/main/org/hornetq/core/client/SessionFailureListener.java =================================================================== --- trunk/src/main/org/hornetq/core/client/SessionFailureListener.java (rev 0) +++ trunk/src/main/org/hornetq/core/client/SessionFailureListener.java 2009-10-30 17:41:19 UTC (rev 8171) @@ -0,0 +1,29 @@ +/* + * 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.client; + +import org.hornetq.core.exception.HornetQException; +import org.hornetq.core.remoting.FailureListener; + +/** + * A SessionFailureListener + * + * @author Tim Fox + * + * + */ +public interface SessionFailureListener extends FailureListener +{ + void beforeReconnect(HornetQException exception); +} Modified: trunk/src/main/org/hornetq/core/client/impl/ClientConsumerImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/client/impl/ClientConsumerImpl.java 2009-10-30 15:12:21 UTC (rev 8170) +++ trunk/src/main/org/hornetq/core/client/impl/ClientConsumerImpl.java 2009-10-30 17:41:19 UTC (rev 8171) @@ -489,7 +489,7 @@ ClientMessageInternal currentChunkMessage = new ClientMessageImpl(packet.getDeliveryCount()); - currentChunkMessage.decodeProperties(ChannelBuffers.wrappedBuffer(packet.getLargeMessageHeader())); + currentChunkMessage.decodeHeadersAndProperties(ChannelBuffers.wrappedBuffer(packet.getLargeMessageHeader())); currentChunkMessage.setLargeMessage(true); Modified: trunk/src/main/org/hornetq/core/client/impl/ClientMessageImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/client/impl/ClientMessageImpl.java 2009-10-30 15:12:21 UTC (rev 8170) +++ trunk/src/main/org/hornetq/core/client/impl/ClientMessageImpl.java 2009-10-30 17:41:19 UTC (rev 8171) @@ -77,6 +77,11 @@ super((byte)0, durable, 0, System.currentTimeMillis(), (byte)4, body); } + public ClientMessageImpl(final boolean durable, final byte[] bytes) + { + super((byte)0, durable, 0, System.currentTimeMillis(), (byte)4, ChannelBuffers.dynamicBuffer(bytes)); + } + public ClientMessageImpl(final boolean durable) { super((byte)0, durable, 0, System.currentTimeMillis(), (byte)4, ChannelBuffers.dynamicBuffer(1024)); Added: trunk/src/main/org/hornetq/core/client/impl/ClientProducerCreditManager.java =================================================================== --- trunk/src/main/org/hornetq/core/client/impl/ClientProducerCreditManager.java (rev 0) +++ trunk/src/main/org/hornetq/core/client/impl/ClientProducerCreditManager.java 2009-10-30 17:41:19 UTC (rev 8171) @@ -0,0 +1,34 @@ +/* + * 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.client.impl; + +import org.hornetq.utils.SimpleString; + +/** + * A ClientProducerCreditManager + * + * @author Tim Fox + * + * + */ +public interface ClientProducerCreditManager +{ + ClientProducerCredits getCredits(SimpleString destination); + + void receiveCredits(SimpleString destination, int credits); + + void reset(); + + void close(); +} Added: trunk/src/main/org/hornetq/core/client/impl/ClientProducerCreditManagerImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/client/impl/ClientProducerCreditManagerImpl.java (rev 0) +++ trunk/src/main/org/hornetq/core/client/impl/ClientProducerCreditManagerImpl.java 2009-10-30 17:41:19 UTC (rev 8171) @@ -0,0 +1,85 @@ +/* + * 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.client.impl; + +import java.util.HashMap; +import java.util.Map; + +import org.hornetq.utils.SimpleString; + +/** + * A ProducerCreditManager + * + * @author <a href="mailto:tim.fox@jboss.com">Tim Fox</a> + * + * + */ +public class ClientProducerCreditManagerImpl implements ClientProducerCreditManager +{ + private final Map<SimpleString, ClientProducerCredits> producerCredits = new HashMap<SimpleString, ClientProducerCredits>(); + + private final ClientSessionInternal session; + + private final int windowSize; + + public ClientProducerCreditManagerImpl(final ClientSessionInternal session, final int windowSize) + { + this.session = session; + + this.windowSize = windowSize; + } + + public synchronized ClientProducerCredits getCredits(final SimpleString destination) + { + ClientProducerCredits credits = producerCredits.get(destination); + + if (credits == null) + { + // Doesn't need to be fair since session is single threaded + credits = new ClientProducerCreditsImpl(session, destination, windowSize); + + producerCredits.put(destination, credits); + } + + return credits; + } + + public synchronized void receiveCredits(final SimpleString destination, final int credits) + { + ClientProducerCredits cr = producerCredits.get(destination); + + if (cr != null) + { + cr.receiveCredits(credits); + } + } + + public synchronized void reset() + { + for (ClientProducerCredits credits : producerCredits.values()) + { + credits.reset(); + } + } + + public synchronized void close() + { + for (ClientProducerCredits credits : producerCredits.values()) + { + credits.close(); + } + + producerCredits.clear(); + } +} Added: trunk/src/main/org/hornetq/core/client/impl/ClientProducerCredits.java =================================================================== --- trunk/src/main/org/hornetq/core/client/impl/ClientProducerCredits.java (rev 0) +++ trunk/src/main/org/hornetq/core/client/impl/ClientProducerCredits.java 2009-10-30 17:41:19 UTC (rev 8171) @@ -0,0 +1,32 @@ +/* + * 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.client.impl; + +/** + * A ClientProducerCredits + * + * @author Tim Fox + * + * + */ +public interface ClientProducerCredits +{ + void acquireCredits(final int credits) throws InterruptedException; + + void receiveCredits(final int credits); + + void reset(); + + void close(); +} Added: trunk/src/main/org/hornetq/core/client/impl/ClientProducerCreditsImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/client/impl/ClientProducerCreditsImpl.java (rev 0) +++ trunk/src/main/org/hornetq/core/client/impl/ClientProducerCreditsImpl.java 2009-10-30 17:41:19 UTC (rev 8171) @@ -0,0 +1,112 @@ +/* + * 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.client.impl; + +import java.util.concurrent.Semaphore; + +import org.hornetq.core.logging.Logger; +import org.hornetq.utils.SimpleString; + +/** + * A ProducerCredits + * + * @author <a href="mailto:tim.fox@jboss.com">Tim Fox</a> + * + * + */ +public class ClientProducerCreditsImpl implements ClientProducerCredits +{ + private static final Logger log = Logger.getLogger(ClientProducerCreditsImpl.class); + + private Semaphore semaphore; + + private final int windowSize; + + private final SimpleString destination; + + private final ClientSessionInternal session; + + private int arriving; + + public ClientProducerCreditsImpl(final ClientSessionInternal session, + final SimpleString destination, + final int windowSize) + { + this.session = session; + + this.destination = destination; + + this.windowSize = windowSize / 2; + + // Doesn't need to be fair since session is single threaded + + semaphore = new Semaphore(0, false); + + // We initial request twice as many credits as we request in subsequent requests + // This allows the producer to keep sending as more arrive, minimising pauses + checkCredits(windowSize); + } + + public void acquireCredits(final int credits) throws InterruptedException + { + checkCredits(credits); + + semaphore.acquire(credits); + } + + public synchronized void receiveCredits(final int credits) + { + arriving -= credits; + + semaphore.release(credits); + } + + public synchronized void reset() + { + //Any arriving credits from before failover won't arrive, so we re-initialise + + semaphore.drainPermits(); + + arriving = 0; + + checkCredits(windowSize * 2); + } + + public void close() + { + //Closing a producer that is blocking should make it return + + semaphore.release(Integer.MAX_VALUE / 2); + } + + private synchronized void checkCredits(final int credits) + { + int needed = Math.max(credits, windowSize); + + if (semaphore.availablePermits() + arriving < needed) + { + int toRequest = needed - arriving; + + arriving += toRequest; + + requestCredits(toRequest); + } + } + + private void requestCredits(final int credits) + { + session.sendProducerCreditsMessage(credits, destination); + } + +} Modified: trunk/src/main/org/hornetq/core/client/impl/ClientProducerImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/client/impl/ClientProducerImpl.java 2009-10-30 15:12:21 UTC (rev 8170) +++ trunk/src/main/org/hornetq/core/client/impl/ClientProducerImpl.java 2009-10-30 17:41:19 UTC (rev 8171) @@ -15,7 +15,6 @@ import static org.hornetq.utils.SimpleString.toSimpleString; -import java.io.BufferedInputStream; import java.io.IOException; import java.io.InputStream; @@ -70,6 +69,8 @@ private final SimpleString groupID; private final int minLargeMessageSize; + + private final ClientProducerCredits credits; // Static --------------------------------------------------------------------------------------- @@ -106,6 +107,15 @@ } this.minLargeMessageSize = minLargeMessageSize; + + if (address != null) + { + credits = session.getCredits(address); + } + else + { + credits = null; + } } // ClientProducer implementation ---------------------------------------------------------------- @@ -116,7 +126,7 @@ } public void send(final Message msg) throws HornetQException - { + { checkClosed(); doSend(null, msg); @@ -191,13 +201,20 @@ private void doSend(final SimpleString address, final Message msg) throws HornetQException { + ClientProducerCredits theCredits; + if (address != null) { msg.setDestination(address); + + //Anonymous + theCredits = session.getCredits(address); } else { msg.setDestination(this.address); + + theCredits = credits; } if (rateLimiter != null) @@ -217,11 +234,11 @@ SessionSendMessage message = new SessionSendMessage(msg, sendBlocking); session.workDone(); - + if (msg.getBodyInputStream() != null || msg.getEncodeSize() >= minLargeMessageSize || msg.isLargeMessage()) { sendMessageInChunks(sendBlocking, msg); - } + } else if (sendBlocking) { channel.sendBlocking(message); @@ -230,6 +247,19 @@ { channel.send(message); } + + try + { + //This will block if credits are not available + + //Note, that for a large message, the encode size only includes the properties + headers + //Not the continuations, but this is ok since we are only interested in limiting the amount of + //data in *memory* and continuations go straight to the disk + theCredits.acquireCredits(msg.getEncodeSize()); + } + catch (InterruptedException e) + { + } } /** @@ -238,7 +268,7 @@ */ private void sendMessageInChunks(final boolean sendBlocking, final Message msg) throws HornetQException { - int headerSize = msg.getPropertiesEncodeSize(); + int headerSize = msg.getHeadersAndPropertiesEncodeSize(); if (headerSize >= minLargeMessageSize) { @@ -253,7 +283,7 @@ } HornetQBuffer headerBuffer = ChannelBuffers.buffer(headerSize); - msg.encodeProperties(headerBuffer); + msg.encodeHeadersAndProperties(headerBuffer); SessionSendLargeMessage initialChunk = new SessionSendLargeMessage(headerBuffer.array()); Modified: trunk/src/main/org/hornetq/core/client/impl/ClientSessionFactoryImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/client/impl/ClientSessionFactoryImpl.java 2009-10-30 15:12:21 UTC (rev 8170) +++ trunk/src/main/org/hornetq/core/client/impl/ClientSessionFactoryImpl.java 2009-10-30 17:41:19 UTC (rev 8171) @@ -73,6 +73,8 @@ public static final int DEFAULT_CONSUMER_MAX_RATE = -1; + 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_MAX_RATE = -1; @@ -98,11 +100,11 @@ public static final long DEFAULT_RETRY_INTERVAL = 2000; public static final double DEFAULT_RETRY_INTERVAL_MULTIPLIER = 1d; - + public static final long DEFAULT_MAX_RETRY_INTERVAL = 2000; public static final int DEFAULT_RECONNECT_ATTEMPTS = 0; - + public static final boolean DEFAULT_FAILOVER_ON_SERVER_SHUTDOWN = false; public static final boolean DEFAULT_USE_GLOBAL_POOLS = true; @@ -158,6 +160,8 @@ private int consumerMaxRate; + private int confirmationWindowSize; + private int producerWindowSize; private int producerMaxRate; @@ -185,15 +189,15 @@ private long retryInterval; private double retryIntervalMultiplier; - + private long maxRetryInterval; private int reconnectAttempts; - + private volatile boolean closed; private boolean failoverOnServerShutdown; - + private final List<Interceptor> interceptors = new CopyOnWriteArrayList<Interceptor>(); private static ExecutorService globalThreadPool; @@ -235,7 +239,7 @@ else { ThreadFactory factory = new HornetQThreadFactory("HornetQ-client-factory-threads-" + System.identityHashCode(this), - true); + true); if (threadPoolMaxSize == -1) { @@ -246,7 +250,8 @@ threadPool = Executors.newFixedThreadPool(threadPoolMaxSize, factory); } - factory = new HornetQThreadFactory("HornetQ-client-factory-pinger-threads-" + System.identityHashCode(this), true); + factory = new HornetQThreadFactory("HornetQ-client-factory-pinger-threads-" + System.identityHashCode(this), + true); scheduledThreadPool = Executors.newScheduledThreadPool(scheduledThreadPoolMaxSize, factory); } @@ -277,19 +282,19 @@ for (Pair<TransportConfiguration, TransportConfiguration> pair : staticConnectors) { FailoverManager cm = new FailoverManagerImpl(this, - pair.a, - pair.b, - failoverOnServerShutdown, - callTimeout, - clientFailureCheckPeriod, - connectionTTL, - retryInterval, - retryIntervalMultiplier, - maxRetryInterval, - reconnectAttempts, - threadPool, - scheduledThreadPool, - interceptors); + pair.a, + pair.b, + failoverOnServerShutdown, + callTimeout, + clientFailureCheckPeriod, + connectionTTL, + retryInterval, + retryIntervalMultiplier, + maxRetryInterval, + reconnectAttempts, + threadPool, + scheduledThreadPool, + interceptors); failoverManagerMap.put(pair, cm); } @@ -324,6 +329,8 @@ consumerMaxRate = DEFAULT_CONSUMER_MAX_RATE; + confirmationWindowSize = DEFAULT_CONFIRMATION_WINDOW_SIZE; + producerWindowSize = DEFAULT_PRODUCER_WINDOW_SIZE; producerMaxRate = DEFAULT_PRODUCER_MAX_RATE; @@ -353,11 +360,11 @@ retryInterval = DEFAULT_RETRY_INTERVAL; retryIntervalMultiplier = DEFAULT_RETRY_INTERVAL_MULTIPLIER; - + maxRetryInterval = DEFAULT_MAX_RETRY_INTERVAL; reconnectAttempts = DEFAULT_RECONNECT_ATTEMPTS; - + failoverOnServerShutdown = DEFAULT_FAILOVER_ON_SERVER_SHUTDOWN; } @@ -483,12 +490,23 @@ this.consumerMaxRate = consumerMaxRate; } + public synchronized int getConfirmationWindowSize() + { + return confirmationWindowSize; + } + + public synchronized void setConfirmationWindowSize(int confirmationWindowSize) + { + checkWrite(); + this.confirmationWindowSize = confirmationWindowSize; + } + public synchronized int getProducerWindowSize() { return producerWindowSize; } - public synchronized void setProducerWindowSize(int producerWindowSize) + public synchronized void setProducerWindowSize(final int producerWindowSize) { checkWrite(); this.producerWindowSize = producerWindowSize; @@ -625,7 +643,7 @@ checkWrite(); this.retryInterval = retryInterval; } - + public synchronized long getMaxRetryInterval() { return maxRetryInterval; @@ -658,7 +676,7 @@ checkWrite(); this.reconnectAttempts = reconnectAttempts; } - + public synchronized boolean isFailoverOnServerShutdown() { return failoverOnServerShutdown; @@ -707,7 +725,7 @@ { return discoveryRefreshTimeout; } - + public void addInterceptor(final Interceptor interceptor) { interceptors.add(interceptor); @@ -740,18 +758,10 @@ preAcknowledge, ackBatchSize); } - - public ClientSession createSession(boolean autoCommitSends, boolean autoCommitAcks, int ackBatchSize) throws HornetQException { - return createSessionInternal(null, - null, - false, - autoCommitSends, - autoCommitAcks, - preAcknowledge, - ackBatchSize); + return createSessionInternal(null, null, false, autoCommitSends, autoCommitAcks, preAcknowledge, ackBatchSize); } public ClientSession createXASession() throws HornetQException @@ -835,7 +845,7 @@ log.error("Failed to stop discovery group", e); } } - + for (FailoverManager failoverManager : failoverManagerMap.values()) { failoverManager.causeExit(); @@ -891,7 +901,7 @@ } Iterator<Map.Entry<Pair<TransportConfiguration, TransportConfiguration>, FailoverManager>> iter = failoverManagerMap.entrySet() - .iterator(); + .iterator(); while (iter.hasNext()) { Map.Entry<Pair<TransportConfiguration, TransportConfiguration>, FailoverManager> entry = iter.next(); @@ -911,19 +921,19 @@ // Create a new failoverManager FailoverManager failoverManager = new FailoverManagerImpl(this, - connectorPair.a, - connectorPair.b, - failoverOnServerShutdown, - callTimeout, - clientFailureCheckPeriod, - connectionTTL, - retryInterval, - retryIntervalMultiplier, - maxRetryInterval, - reconnectAttempts, - threadPool, - scheduledThreadPool, - interceptors); + connectorPair.a, + connectorPair.b, + failoverOnServerShutdown, + callTimeout, + clientFailureCheckPeriod, + connectionTTL, + retryInterval, + retryIntervalMultiplier, + maxRetryInterval, + reconnectAttempts, + threadPool, + scheduledThreadPool, + interceptors); failoverManagerMap.put(connectorPair, failoverManager); } @@ -991,7 +1001,7 @@ if (!ok) { throw new HornetQException(HornetQException.CONNECTION_TIMEDOUT, - "Timed out waiting to receive initial broadcast from discovery group"); + "Timed out waiting to receive initial broadcast from discovery group"); } } @@ -1002,22 +1012,23 @@ FailoverManager failoverManager = failoverManagerArray[pos]; ClientSession session = failoverManager.createSession(username, - password, - xa, - autoCommitSends, - autoCommitAcks, - preAcknowledge, - ackBatchSize, - cacheLargeMessagesClient, - minLargeMessageSize, - blockOnAcknowledge, - autoGroup, - producerWindowSize, - consumerWindowSize, - producerMaxRate, - consumerMaxRate, - blockOnNonPersistentSend, - blockOnPersistentSend); + password, + xa, + autoCommitSends, + autoCommitAcks, + preAcknowledge, + ackBatchSize, + cacheLargeMessagesClient, + minLargeMessageSize, + blockOnAcknowledge, + autoGroup, + confirmationWindowSize, + producerWindowSize, + consumerWindowSize, + producerMaxRate, + consumerMaxRate, + blockOnNonPersistentSend, + blockOnPersistentSend); return session; } Modified: trunk/src/main/org/hornetq/core/client/impl/ClientSessionImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/client/impl/ClientSessionImpl.java 2009-10-30 15:12:21 UTC (rev 8170) +++ trunk/src/main/org/hornetq/core/client/impl/ClientSessionImpl.java 2009-10-30 17:41:19 UTC (rev 8171) @@ -31,6 +31,7 @@ import org.hornetq.core.client.ClientMessage; import org.hornetq.core.client.ClientProducer; import org.hornetq.core.client.SendAcknowledgementHandler; +import org.hornetq.core.client.SessionFailureListener; import org.hornetq.core.exception.HornetQException; import org.hornetq.core.logging.Logger; import org.hornetq.core.remoting.Channel; @@ -57,6 +58,7 @@ import org.hornetq.core.remoting.impl.wireformat.SessionQueueQueryResponseMessage; import org.hornetq.core.remoting.impl.wireformat.SessionReceiveContinuationMessage; import org.hornetq.core.remoting.impl.wireformat.SessionReceiveMessage; +import org.hornetq.core.remoting.impl.wireformat.SessionRequestProducerCreditsMessage; import org.hornetq.core.remoting.impl.wireformat.SessionSendMessage; import org.hornetq.core.remoting.impl.wireformat.SessionXACommitMessage; import org.hornetq.core.remoting.impl.wireformat.SessionXAEndMessage; @@ -107,7 +109,7 @@ // Attributes ---------------------------------------------------------------------------- - private final FailoverManager connectionManager; + private final FailoverManager failoverManager; private final String name; @@ -144,9 +146,11 @@ private final int consumerMaxRate; - private final int producerWindowSize; + private final int confirmationWindowSize; private final int producerMaxRate; + + private final int producerWindowSize; private final boolean blockOnNonPersistentSend; @@ -165,12 +169,12 @@ private final IDGenerator idGenerator = new SimpleIDGenerator(0); + private final ClientProducerCreditManager producerCreditManager; + private volatile boolean started; private SendAcknowledgementHandler sendAckHandler; - private volatile boolean closedSent; - private volatile boolean rollbackOnly; private volatile boolean workDone; @@ -190,6 +194,7 @@ final int ackBatchSize, final int consumerWindowSize, final int consumerMaxRate, + final int confirmationWindowSize, final int producerWindowSize, final int producerMaxRate, final boolean blockOnNonPersistentSend, @@ -201,7 +206,7 @@ final Channel channel, final Executor executor) throws HornetQException { - this.connectionManager = connectionManager; + this.failoverManager = connectionManager; this.name = name; @@ -235,8 +240,10 @@ this.consumerMaxRate = consumerMaxRate; + this.confirmationWindowSize = confirmationWindowSize; + this.producerWindowSize = producerWindowSize; - + this.producerMaxRate = producerMaxRate; this.blockOnNonPersistentSend = blockOnNonPersistentSend; @@ -246,6 +253,8 @@ this.cacheLargeMessageClient = cacheLargeMessageClient; this.minLargeMessageSize = minLargeMessageSize; + + producerCreditManager = new ClientProducerCreditManagerImpl(this, producerWindowSize); } // ClientSession implementation @@ -329,7 +338,7 @@ return response; } - + public void forceDelivery(long consumerID, long sequence) throws HornetQException { checkClosed(); @@ -544,14 +553,21 @@ return new ClientMessageImpl(durable, body); } - - /* (non-Javadoc) - * @see org.hornetq.core.client.impl.ClientSessionInternal#createBuffer(int) - */ + + public ClientMessage createClientMessage(final boolean durable, final HornetQBuffer buffer) + { + return new ClientMessageImpl(durable, buffer); + } + public HornetQBuffer createBuffer(final int size) { return ChannelBuffers.dynamicBuffer(size); } + + public HornetQBuffer createBuffer(final byte[] bytes) + { + return ChannelBuffers.dynamicBuffer(bytes); + } public boolean isClosed() { @@ -612,14 +628,14 @@ } } - public void addFailureListener(final FailureListener listener) + public void addFailureListener(final SessionFailureListener listener) { - connectionManager.addFailureListener(listener); + failoverManager.addFailureListener(listener); } - public boolean removeFailureListener(final FailureListener listener) + public boolean removeFailureListener(final SessionFailureListener listener) { - return connectionManager.removeFailureListener(listener); + return failoverManager.removeFailureListener(listener); } public int getVersion() @@ -756,10 +772,10 @@ try { + producerCreditManager.close(); + closeChildren(); - closedSent = true; - channel.sendBlocking(new SessionCloseMessage()); } catch (Throwable ignore) @@ -776,7 +792,9 @@ { return; } - + + producerCreditManager.close(); + cleanUpChildren(); doCleanup(); @@ -798,6 +816,8 @@ return; } + boolean resetCreditManager = false; + // We lock the channel to prevent any packets to be added to the resend // cache during the failover process channel.lock(); @@ -812,7 +832,7 @@ Packet request = new ReattachSessionMessage(name, channel.getLastReceivedCommandID()); - Channel channel1 = backupConnection.getChannel(1, -1, false); + Channel channel1 = backupConnection.getChannel(1, -1); ReattachSessionResponseMessage response = (ReattachSessionResponseMessage)channel1.sendBlocking(request); @@ -828,7 +848,7 @@ // session // won't exist or the target server has been restarted - in this case the session will need to be recreated, // and we'll need to recreate any consumers - + Packet createRequest = new CreateSessionMessage(name, channel.getID(), version, @@ -839,12 +859,12 @@ autoCommitSends, autoCommitAcks, preAcknowledge, - producerWindowSize); + confirmationWindowSize); channel1.sendBlocking(createRequest); channel.clearCommands(); - + for (Map.Entry<Long, ClientConsumerInternal> entry : consumers.entrySet()) { SessionCreateConsumerMessage createConsumerRequest = new SessionCreateConsumerMessage(entry.getKey(), @@ -911,10 +931,11 @@ conn.write(buffer, false); } - + + resetCreditManager = true; + channel.returnBlocking(); - } - + } } catch (Throwable t) { @@ -924,6 +945,11 @@ { channel.unlock(); } + + if (resetCreditManager) + { + producerCreditManager.reset(); + } } public void workDone() @@ -938,9 +964,24 @@ public FailoverManager getConnectionManager() { - return connectionManager; + return failoverManager; } + public void sendProducerCreditsMessage(final int credits, final SimpleString destination) + { + channel.send(new SessionRequestProducerCreditsMessage(credits, destination)); + } + + public ClientProducerCredits getCredits(final SimpleString address) + { + return producerCreditManager.getCredits(address); + } + + public void handleReceiveProducerCredits(final SimpleString address, final int credits) + { + producerCreditManager.receiveCredits(address, credits); + } + // CommandConfirmationHandler implementation ------------------------------------ public void commandConfirmed(final Packet packet) @@ -1087,7 +1128,7 @@ ClientSessionInternal other = (ClientSessionInternal)xares; - return connectionManager == other.getConnectionManager(); + return failoverManager == other.getConnectionManager(); } public int prepare(final Xid xid) throws XAException @@ -1256,7 +1297,7 @@ // FailureListener implementation -------------------------------------------- public void connectionFailed(final HornetQException me) - { + { try { cleanUp(); @@ -1288,7 +1329,7 @@ // Private // ---------------------------------------------------------------------------- - + private int calcWindowSize(final int windowSize) { int clientWindowSize; @@ -1386,7 +1427,7 @@ final boolean blockOnPersistentSend) throws HornetQException { checkClosed(); - + ClientProducerInternal producer = new ClientProducerImpl(this, address, maxRate == -1 ? null @@ -1438,7 +1479,7 @@ } private void doCleanup() - { + { remotingConnection.removeFailureListener(this); synchronized (this) @@ -1448,7 +1489,7 @@ channel.close(); } - connectionManager.removeSession(this); + failoverManager.removeSession(this); } private void cleanUpChildren() throws Exception Modified: trunk/src/main/org/hornetq/core/client/impl/ClientSessionInternal.java =================================================================== --- trunk/src/main/org/hornetq/core/client/impl/ClientSessionInternal.java 2009-10-30 15:12:21 UTC (rev 8170) +++ trunk/src/main/org/hornetq/core/client/impl/ClientSessionInternal.java 2009-10-30 17:41:19 UTC (rev 8171) @@ -18,7 +18,6 @@ import org.hornetq.core.remoting.RemotingConnection; import org.hornetq.core.remoting.impl.wireformat.SessionReceiveContinuationMessage; import org.hornetq.core.remoting.impl.wireformat.SessionReceiveMessage; -import org.hornetq.core.remoting.spi.HornetQBuffer; import org.hornetq.utils.SimpleString; /** @@ -36,8 +35,6 @@ int getMinLargeMessageSize(); - HornetQBuffer createBuffer(int size); - void expire(long consumerID, long messageID) throws HornetQException; void addConsumer(ClientConsumerInternal consumer); @@ -70,5 +67,9 @@ void forceDelivery(long consumerID, long sequence) throws HornetQException; - //void sendProducerCreditsMessage(int credits, SimpleString destination); + void sendProducerCreditsMessage(int credits, SimpleString destination); + + ClientProducerCredits getCredits(SimpleString address); + + void handleReceiveProducerCredits(SimpleString address, int credits); } Modified: trunk/src/main/org/hornetq/core/client/impl/ClientSessionPacketHandler.java =================================================================== --- trunk/src/main/org/hornetq/core/client/impl/ClientSessionPacketHandler.java 2009-10-30 15:12:21 UTC (rev 8170) +++ trunk/src/main/org/hornetq/core/client/impl/ClientSessionPacketHandler.java 2009-10-30 17:41:19 UTC (rev 8171) @@ -22,6 +22,8 @@ import org.hornetq.core.remoting.ChannelHandler; import org.hornetq.core.remoting.Packet; import org.hornetq.core.remoting.impl.wireformat.HornetQExceptionMessage; +import org.hornetq.core.remoting.impl.wireformat.PacketImpl; +import org.hornetq.core.remoting.impl.wireformat.SessionProducerCreditsMessage; import org.hornetq.core.remoting.impl.wireformat.SessionReceiveContinuationMessage; import org.hornetq.core.remoting.impl.wireformat.SessionReceiveMessage; @@ -78,6 +80,14 @@ break; } + case PacketImpl.SESS_PRODUCER_CREDITS: + { + SessionProducerCreditsMessage message = (SessionProducerCreditsMessage)packet; + + clientSession.handleReceiveProducerCredits(message.getAddress(), message.getCredits()); + + break; + } case EXCEPTION: { //TODO - we can provide a means for async exceptions to get back to to client Modified: trunk/src/main/org/hornetq/core/client/impl/DelegatingSession.java =================================================================== --- trunk/src/main/org/hornetq/core/client/impl/DelegatingSession.java 2009-10-30 15:12:21 UTC (rev 8170) +++ trunk/src/main/org/hornetq/core/client/impl/DelegatingSession.java 2009-10-30 17:41:19 UTC (rev 8171) @@ -23,9 +23,9 @@ import org.hornetq.core.client.ClientMessage; import org.hornetq.core.client.ClientProducer; import org.hornetq.core.client.SendAcknowledgementHandler; +import org.hornetq.core.client.SessionFailureListener; import org.hornetq.core.exception.HornetQException; import org.hornetq.core.logging.Logger; -import org.hornetq.core.remoting.FailureListener; import org.hornetq.core.remoting.RemotingConnection; import org.hornetq.core.remoting.impl.wireformat.SessionBindingQueryResponseMessage; import org.hornetq.core.remoting.impl.wireformat.SessionQueueQueryResponseMessage; @@ -104,7 +104,7 @@ session.addConsumer(consumer); } - public void addFailureListener(FailureListener listener) + public void addFailureListener(SessionFailureListener listener) { session.addFailureListener(listener); } @@ -149,11 +149,6 @@ session.commit(xid, onePhase); } - public HornetQBuffer createBuffer(int size) - { - return session.createBuffer(size); - } - public ClientMessage createClientMessage(boolean durable) { return session.createClientMessage(durable); @@ -168,7 +163,23 @@ { return session.createClientMessage(type, durable); } + + public ClientMessage createClientMessage(boolean durable, HornetQBuffer buffer) + { + return session.createClientMessage(durable, buffer); + } + + public HornetQBuffer createBuffer(byte[] bytes) + { + return session.createBuffer(bytes); + } + public HornetQBuffer createBuffer(int size) + { + return session.createBuffer(size); + } + + public ClientConsumer createConsumer(SimpleString queueName, SimpleString filterString, boolean browseOnly) throws HornetQException { return session.createConsumer(queueName, filterString, browseOnly); @@ -443,7 +454,7 @@ session.removeConsumer(consumer); } - public boolean removeFailureListener(FailureListener listener) + public boolean removeFailureListener(SessionFailureListener listener) { return session.removeFailureListener(listener); } @@ -517,4 +528,19 @@ { session.workDone(); } + + public void sendProducerCreditsMessage(int credits, SimpleString destination) + { + session.sendProducerCreditsMessage(credits, destination); + } + + public ClientProducerCredits getCredits(SimpleString address) + { + return session.getCredits(address); + } + + public void handleReceiveProducerCredits(SimpleString address, int credits) + { + session.handleReceiveProducerCredits(address, credits); + } } Modified: trunk/src/main/org/hornetq/core/client/impl/FailoverManager.java =================================================================== --- trunk/src/main/org/hornetq/core/client/impl/FailoverManager.java 2009-10-30 15:12:21 UTC (rev 8170) +++ trunk/src/main/org/hornetq/core/client/impl/FailoverManager.java 2009-10-30 17:41:19 UTC (rev 8171) @@ -14,8 +14,8 @@ package org.hornetq.core.client.impl; import org.hornetq.core.client.ClientSession; +import org.hornetq.core.client.SessionFailureListener; import org.hornetq.core.exception.HornetQException; -import org.hornetq.core.remoting.FailureListener; import org.hornetq.core.remoting.RemotingConnection; /** @@ -40,6 +40,7 @@ final int minLargeMessageSize, final boolean blockOnAcknowledge, final boolean autoGroup, + final int confirmationWindowSize, final int producerWindowSize, final int consumerWindowSize, final int producerMaxRate, @@ -55,9 +56,9 @@ int numSessions(); - void addFailureListener(FailureListener listener); + void addFailureListener(SessionFailureListener listener); - boolean removeFailureListener(FailureListener listener); + boolean removeFailureListener(SessionFailureListener listener); void causeExit(); } Modified: trunk/src/main/org/hornetq/core/client/impl/FailoverManagerImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/client/impl/FailoverManagerImpl.java 2009-10-30 15:12:21 UTC (rev 8170) +++ trunk/src/main/org/hornetq/core/client/impl/FailoverManagerImpl.java 2009-10-30 17:41:19 UTC (rev 8171) @@ -27,6 +27,7 @@ import org.hornetq.core.client.ClientSession; import org.hornetq.core.client.ClientSessionFactory; +import org.hornetq.core.client.SessionFailureListener; import org.hornetq.core.config.TransportConfiguration; import org.hornetq.core.exception.HornetQException; import org.hornetq.core.logging.Logger; @@ -124,7 +125,7 @@ private boolean failoverOnServerShutdown; - private Set<FailureListener> listeners = new ConcurrentHashSet<FailureListener>(); + private Set<SessionFailureListener> listeners = new ConcurrentHashSet<SessionFailureListener>(); private Connector connector; @@ -239,6 +240,7 @@ final int minLargeMessageSize, final boolean blockOnAcknowledge, final boolean autoGroup, + final int confWindowSize, final int producerWindowSize, final int consumerWindowSize, final int producerMaxRate, @@ -280,7 +282,7 @@ } - channel1 = connection.getChannel(1, -1, false); + channel1 = connection.getChannel(1, -1); // Lock it - this must be done while the failoverLock is held channel1.getLock().lock(); @@ -306,7 +308,7 @@ autoCommitSends, autoCommitAcks, preAcknowledge, - producerWindowSize); + confWindowSize); Packet pResponse; try @@ -337,8 +339,7 @@ CreateSessionResponseMessage response = (CreateSessionResponseMessage)pResponse; Channel sessionChannel = connection.getChannel(sessionChannelID, - producerWindowSize, - producerWindowSize != -1); + confWindowSize); ClientSessionInternal session = new ClientSessionImpl(this, name, @@ -353,6 +354,7 @@ ackBatchSize, consumerWindowSize, consumerMaxRate, + confWindowSize, producerWindowSize, producerMaxRate, blockOnNonPersistentSend, @@ -446,12 +448,12 @@ return sessions.size(); } - public void addFailureListener(FailureListener listener) + public void addFailureListener(final SessionFailureListener listener) { listeners.add(listener); } - public boolean removeFailureListener(FailureListener listener) + public boolean removeFailureListener(final SessionFailureListener listener) { return listeners.remove(listener); } @@ -497,6 +499,9 @@ // listeners again return; } + + //We call before reconnection occurs to give the user a chance to do cleanup, like cancel messages + callFailureListeners(me, false); // Now get locks on all channel 1s, whilst holding the failoverLock - this makes sure // There are either no threads executing in createSession, or one is blocking on a createSession @@ -539,7 +544,7 @@ { attemptReconnect = reconnectAttempts != 0; } - + if (attemptFailover || attemptReconnect) { lockChannel1(); @@ -622,22 +627,28 @@ connection.destroy(); connection = null; - } - - // We always call the failure listeners - callFailureListeners(me); + } + + callFailureListeners(me, true); } } - private void callFailureListeners(final HornetQException me) + private void callFailureListeners(final HornetQException me, final boolean afterReconnect) { - final List<FailureListener> listenersClone = new ArrayList<FailureListener>(listeners); + final List<SessionFailureListener> listenersClone = new ArrayList<SessionFailureListener>(listeners); - for (final FailureListener listener : listenersClone) + for (final SessionFailureListener listener : listenersClone) { try { - listener.connectionFailed(me); + if (afterReconnect) + { + listener.connectionFailed(me); + } + else + { + listener.beforeReconnect(me); + } } catch (final Throwable t) { @@ -653,9 +664,9 @@ * Re-attach sessions all pre-existing sessions to the new remoting connection */ private void reconnectSessions(final int reconnectAttempts) - { + { RemotingConnection backupConnection = getConnectionWithRetry(reconnectAttempts); - + if (backupConnection == null) { log.warn("Failed to connect to server."); @@ -722,7 +733,7 @@ catch (InterruptedException ignore) { } - + // Exponential back-off long newInterval = (long)((double)interval * retryIntervalMultiplier); @@ -870,7 +881,7 @@ connection.addFailureListener(new DelegatingFailureListener(connection.getID())); - connection.getChannel(0, -1, false).setHandler(new Channel0Handler(connection)); + connection.getChannel(0, -1).setHandler(new Channel0Handler(connection)); if (clientFailureCheckPeriod != -1) { @@ -912,21 +923,21 @@ private void lockChannel1() { - Channel channel1 = connection.getChannel(1, -1, false); + Channel channel1 = connection.getChannel(1, -1); channel1.getLock().lock(); } private void unlockChannel1() { - Channel channel1 = connection.getChannel(1, -1, false); + Channel channel1 = connection.getChannel(1, -1); channel1.getLock().unlock(); } private void forceReturnChannel1() { - Channel channel1 = connection.getChannel(1, -1, false); + Channel channel1 = connection.getChannel(1, -1); channel1.returnBlocking(); } @@ -981,8 +992,6 @@ { public void bufferReceived(final Object connectionID, final HornetQBuffer buffer) { - // ConnectionEntry entry = connections.get(connectionID); - if (connection != null && connectionID == connection.getID()) { connection.bufferReceived(connectionID, buffer); @@ -1073,7 +1082,7 @@ Ping ping = new Ping(connectionTTL); - Channel channel0 = connection.getChannel(0, -1, false); + Channel channel0 = connection.getChannel(0, -1); channel0.send(ping); } Modified: trunk/src/main/org/hornetq/core/config/impl/Validators.java =================================================================== --- trunk/src/main/org/hornetq/core/config/impl/Validators.java 2009-10-30 15:12:21 UTC (rev 8170) +++ trunk/src/main/org/hornetq/core/config/impl/Validators.java 2009-10-30 17:41:19 UTC (rev 8171) @@ -16,6 +16,7 @@ import static java.lang.String.format; import org.hornetq.core.server.JournalType; +import org.hornetq.core.settings.impl.AddressFullMessagePolicy; /** * A Validators @@ -32,12 +33,12 @@ // Attributes ---------------------------------------------------- // Static -------------------------------------------------------- - + public static interface Validator { void validate(String name, Object value); } - + public static Validator NO_CHECK = new Validator() { public void validate(String name, Object value) @@ -45,7 +46,7 @@ return; } }; - + public static Validator NOT_NULL_OR_EMPTY = new Validator() { public void validate(String name, Object value) @@ -79,9 +80,11 @@ public void validate(String name, Object value) { Number val = (Number)value; - if (val != null && val.intValue()< 0 || val.intValue() > 100) + if (val != null && val.intValue() < 0 || val.intValue() > 100) { - throw new IllegalArgumentException(format("%s must be a valid percentual value between 0 and 100 (actual value: %s)", name, val)); + throw new IllegalArgumentException(format("%s must be a valid percentual value between 0 and 100 (actual value: %s)", + name, + val)); } } }; @@ -172,6 +175,20 @@ } }; + public static final Validator ADDRESS_FULL_MESSAGE_POLICY_TYPE = new Validator() + { + public void validate(String name, Object value) + { + String val = (String)value; + if (val == null || !val.equals(AddressFullMessagePolicy.PAGE.toString()) && + !val.equals(AddressFullMessagePolicy.DROP.toString()) && + !val.equals(AddressFullMessagePolicy.BLOCK.toString())) + { + throw new IllegalArgumentException("Invalid address full message policy type " + val); + } + } + }; + // Constructors -------------------------------------------------- // Public -------------------------------------------------------- Modified: trunk/src/main/org/hornetq/core/deployers/impl/AddressSettingsDeployer.java =================================================================== --- trunk/src/main/org/hornetq/core/deployers/impl/AddressSettingsDeployer.java 2009-10-30 15:12:21 UTC (rev 8170) +++ trunk/src/main/org/hornetq/core/deployers/impl/AddressSettingsDeployer.java 2009-10-30 17:41:19 UTC (rev 8171) @@ -13,9 +13,11 @@ package org.hornetq.core.deployers.impl; +import org.hornetq.core.config.impl.Validators; import org.hornetq.core.deployers.DeploymentManager; import org.hornetq.core.logging.Logger; import org.hornetq.core.settings.HierarchicalRepository; +import org.hornetq.core.settings.impl.AddressFullMessagePolicy; import org.hornetq.core.settings.impl.AddressSettings; import org.hornetq.utils.SimpleString; import org.w3c.dom.Node; @@ -39,7 +41,7 @@ private static final String MAX_SIZE_BYTES_NODE_NAME = "max-size-bytes"; - private static final String DROP_MESSAGES_WHEN_FULL_NODE_NAME = "drop-messages-when-full"; + private static final String ADDRESS_FULL_MESSAGE_POLICY_NODE_NAME = "address-full-policy"; private static final String PAGE_SIZE_BYTES_NODE_NAME = "page-size-bytes"; @@ -124,9 +126,24 @@ { addressSettings.setMessageCounterHistoryDayLimit(Integer.valueOf(child.getTextContent())); } - else if (DROP_MESSAGES_WHEN_FULL_NODE_NAME.equalsIgnoreCase(child.getNodeName())) + else if (ADDRESS_FULL_MESSAGE_POLICY_NODE_NAME.equalsIgnoreCase(child.getNodeName())) { - addressSettings.setDropMessagesWhenFull(Boolean.valueOf(child.getTextContent().trim())); + String value = child.getTextContent().trim(); + Validators.ADDRESS_FULL_MESSAGE_POLICY_TYPE.validate(ADDRESS_FULL_MESSAGE_POLICY_NODE_NAME, value); + AddressFullMessagePolicy policy = null; + if (value.equals(AddressFullMessagePolicy.BLOCK.toString())) + { + policy = AddressFullMessagePolicy.BLOCK; + } + else if (value.equals(AddressFullMessagePolicy.DROP.toString())) + { + policy = AddressFullMessagePolicy.DROP; + } + else if (value.equals(AddressFullMessagePolicy.PAGE.toString())) + { + policy = AddressFullMessagePolicy.PAGE; + } + addressSettings.setAddressFullMessagePolicy(policy); } else if (LVQ_NODE_NAME.equalsIgnoreCase(child.getNodeName())) { Modified: trunk/src/main/org/hornetq/core/journal/impl/NIOSequentialFile.java =================================================================== --- trunk/src/main/org/hornetq/core/journal/impl/NIOSequentialFile.java 2009-10-30 15:12:21 UTC (rev 8170) +++ trunk/src/main/org/hornetq/core/journal/impl/NIOSequentialFile.java 2009-10-30 17:41:19 UTC (rev 8171) @@ -20,7 +20,6 @@ import java.util.concurrent.atomic.AtomicLong; import org.hornetq.core.journal.IOCallback; -import org.hornetq.core.journal.SequentialFile; import org.hornetq.core.logging.Logger; import org.hornetq.core.remoting.spi.HornetQBuffer; Modified: trunk/src/main/org/hornetq/core/journal/impl/SyncSpeedTest.java =================================================================== --- trunk/src/main/org/hornetq/core/journal/impl/SyncSpeedTest.java 2009-10-30 15:12:21 UTC (rev 8170) +++ trunk/src/main/org/hornetq/core/journal/impl/SyncSpeedTest.java 2009-10-30 17:41:19 UTC (rev 8171) @@ -16,6 +16,7 @@ import java.io.File; import java.io.RandomAccessFile; import java.nio.ByteBuffer; +import java.nio.MappedByteBuffer; import java.nio.channels.FileChannel; import org.hornetq.core.logging.Logger; @@ -45,15 +46,13 @@ e.printStackTrace(); } } - + public void run() throws Exception - { - log.info("******* Starting file sync speed test *******"); + { + int fileSize = 1024 * 1024 * 100; - int fileSize = 1024 * 1024 * 10; + int recordSize = 10 * 1024; - int recordSize = 1024; - int its = 10 * 1024; File file = new File("sync-speed-test.dat"); @@ -75,31 +74,91 @@ channel.position(0); + MappedByteBuffer mappedBB = channel.map(FileChannel.MapMode.READ_WRITE, 0, fileSize); + + mappedBB.load(); + // mappedBB.order(java.nio.ByteOrder.LITTLE_ENDIAN); + System.out.println("isLoaded=" + mappedBB.isLoaded() + "; isDirect=" + mappedBB.isDirect() + "; byteOrder=" + mappedBB.order()); + ByteBuffer bb1 = generateBuffer(recordSize, (byte)'h'); - log.info("Measuring"); + System.out.println("Measuring"); long start = System.currentTimeMillis(); for (int i = 0; i < its; i++) { - write(bb1, channel, recordSize); - - channel.force(false); + bb1.flip(); + mappedBB.position(0); + mappedBB.put(bb1); + mappedBB.force(); + + //write(bb1, channel, recordSize); + // channel.force(false); } long end = System.currentTimeMillis(); double rate = 1000 * ((double)its) / (end - start); - log.info("Rate of " + rate + " syncs per sec"); - - rfile.close(); - + System.out.println("Rate of " + rate + " syncs per sec"); file.delete(); - - log.info("****** test complete *****"); } + +// public void run() throws Exception +// { +// log.info("******* Starting file sync speed test *******"); +// +// int fileSize = 1024 * 1024 * 10; +// +// int recordSize = 1024; +// +// int its = 10 * 1024; +// +// File file = new File("sync-speed-test.dat"); +// +// if (file.exists()) +// { +// file.delete(); +// } +// +// RandomAccessFile rfile = new RandomAccessFile(file, "rw"); +// +// FileChannel channel = rfile.getChannel(); +// +// ByteBuffer bb = generateBuffer(fileSize, (byte)'x'); +// +// write(bb, channel, fileSize); +// +// channel.force(false); +// +// channel.position(0); +// +// ByteBuffer bb1 = generateBuffer(recordSize, (byte)'h'); +// +// log.info("Measuring"); +// +// long start = System.currentTimeMillis(); +// +// for (int i = 0; i < its; i++) +// { +// write(bb1, channel, recordSize); +// +// channel.force(false); +// } +// +// long end = System.currentTimeMillis(); +// +// double rate = 1000 * ((double)its) / (end - start); +// +// log.info("Rate of " + rate + " syncs per sec"); +// +// rfile.close(); +// +// file.delete(); +// +// log.info("****** test complete *****"); +// } private void write(final ByteBuffer buffer, final FileChannel channel, final int size) throws Exception { Modified: trunk/src/main/org/hornetq/core/message/Message.java =================================================================== --- trunk/src/main/org/hornetq/core/message/Message.java 2009-10-30 15:12:21 UTC (rev 8170) +++ trunk/src/main/org/hornetq/core/message/Message.java 2009-10-30 17:41:19 UTC (rev 8171) @@ -68,11 +68,11 @@ void decode(HornetQBuffer buffer); - int getPropertiesEncodeSize(); + int getHeadersAndPropertiesEncodeSize(); - void encodeProperties(HornetQBuffer buffer); + void encodeHeadersAndProperties(HornetQBuffer buffer); - void decodeProperties(HornetQBuffer buffer); + void decodeHeadersAndProperties(HornetQBuffer buffer); int getBodySize(); Modified: trunk/src/main/org/hornetq/core/message/impl/MessageImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/message/impl/MessageImpl.java 2009-10-30 15:12:21 UTC (rev 8170) +++ trunk/src/main/org/hornetq/core/message/impl/MessageImpl.java 2009-10-30 17:41:19 UTC (rev 8171) @@ -169,17 +169,17 @@ public void encode(final HornetQBuffer buffer) { - encodeProperties(buffer); + encodeHeadersAndProperties(buffer); buffer.writeInt(getBodySize()); encodeBody(buffer); } public int getEncodeSize() { - return getPropertiesEncodeSize() + SIZE_INT + getBodySize(); + return getHeadersAndPropertiesEncodeSize() + SIZE_INT + getBodySize(); } - public int getPropertiesEncodeSize() + public int getHeadersAndPropertiesEncodeSize() { return SIZE_LONG + /* Destination */SimpleString.sizeofString(destination) + /* Type */SIZE_BYTE + @@ -195,7 +195,7 @@ return body.writerIndex(); } - public void encodeProperties(HornetQBuffer buffer) + public void encodeHeadersAndProperties(HornetQBuffer buffer) { buffer.writeLong(messageID); buffer.writeSimpleString(destination); @@ -221,12 +221,12 @@ public void decode(final HornetQBuffer buffer) { - decodeProperties(buffer); + decodeHeadersAndProperties(buffer); decodeBody(buffer); } - public void decodeProperties(final HornetQBuffer buffer) + public void decodeHeadersAndProperties(final HornetQBuffer buffer) { messageID = buffer.readLong(); destination = buffer.readSimpleString(); Modified: trunk/src/main/org/hornetq/core/paging/PagingStore.java =================================================================== --- trunk/src/main/org/hornetq/core/paging/PagingStore.java 2009-10-30 15:12:21 UTC (rev 8170) +++ trunk/src/main/org/hornetq/core/paging/PagingStore.java 2009-10-30 17:41:19 UTC (rev 8171) @@ -14,6 +14,10 @@ package org.hornetq.core.paging; 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; import org.hornetq.utils.SimpleString; /** @@ -36,7 +40,7 @@ /** Maximum number of bytes allowed in memory */ long getMaxSizeBytes(); - boolean isDropWhenMaxSize(); + AddressFullMessagePolicy getAddressFullMessagePolicy(); long getPageSizeBytes(); @@ -65,9 +69,35 @@ boolean startDepaging(); /** - * @param memoryEstimate + * + * @param message + * @throws Exception + */ + void addSize(ServerMessage message, boolean add) throws Exception; + + /** + * + * @param reference + * @throws Exception + */ + void addSize(MessageReference reference, boolean add) throws Exception; + + /** + * + * @param desired * @return - * @throws Exception */ - void addSize(long memoryEstimate) throws Exception; + int getAvailableProducerCredits(int desired); + + /** + * + * @param credits + */ + void returnProducerCredits(int credits); + + /** + * + * @return + */ + ServerProducerCreditManager getProducerCreditManager(); } Modified: trunk/src/main/org/hornetq/core/paging/impl/PagingStoreImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/paging/impl/PagingStoreImpl.java 2009-10-30 15:12:21 UTC (rev 8170) +++ trunk/src/main/org/hornetq/core/paging/impl/PagingStoreImpl.java 2009-10-30 17:41:19 UTC (rev 8171) @@ -38,9 +38,12 @@ import org.hornetq.core.persistence.StorageManager; import org.hornetq.core.postoffice.PostOffice; import org.hornetq.core.server.LargeServerMessage; -import org.hornetq.core.server.RoutingContext; +import org.hornetq.core.server.MessageReference; import org.hornetq.core.server.ServerMessage; import org.hornetq.core.server.impl.RoutingContextImpl; +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; import org.hornetq.core.transaction.TransactionPropertyIndexes; @@ -52,6 +55,7 @@ * @see PagingStore * * @author <a href="mailto:clebert.suconic@jboss.com">Clebert Suconic</a> + * @author <a href="mailto:tim.fox@jboss.com">Tim Fox</a> * */ public class PagingStoreImpl implements TestSupportPageStore @@ -80,7 +84,7 @@ private final long pageSize; - private final boolean dropMessagesWhenFull; + private final AddressFullMessagePolicy addressFullMessagePolicy; private boolean printedDropMessagesWarning; @@ -107,9 +111,15 @@ * We need to perform checks on currentPage with minimal locking * */ 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); + // Static -------------------------------------------------------- private static final boolean isTrace = log.isTraceEnabled(); @@ -148,7 +158,7 @@ pageSize = addressSettings.getPageSizeBytes(); - dropMessagesWhenFull = addressSettings.isDropMessagesWhenFull(); + this.addressFullMessagePolicy = addressSettings.getAddressFullMessagePolicy(); this.executor = executor; @@ -157,6 +167,10 @@ this.fileFactory = fileFactory; this.storeFactory = storeFactory; + + this.availableProducerCredits.set(maxSize); + + this.creditManager = new ServerProducerCreditManagerImpl(this); } // Public -------------------------------------------------------- @@ -174,9 +188,9 @@ return maxSize; } - public boolean isDropWhenMaxSize() + public AddressFullMessagePolicy getAddressFullMessagePolicy() { - return dropMessagesWhenFull; + return addressFullMessagePolicy; } public long getPageSizeBytes() @@ -189,9 +203,9 @@ currentPageLock.readLock().lock(); try { - if (isDropWhenMaxSize()) + if (addressFullMessagePolicy != AddressFullMessagePolicy.PAGE) { - return isDrop(); + return isFull(); } else { @@ -213,12 +227,104 @@ { return storeName; } - - public void addSize(final long size) throws Exception + + public boolean isExceededAvailableCredits() { - if (isDropWhenMaxSize()) + 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); + } + + private synchronized void checkReleaseProducerFlowControlCredits(final long size) + { + if (addressFullMessagePolicy == AddressFullMessagePolicy.BLOCK && maxSize != -1) + { + long avail = availableProducerCredits.addAndGet(-size); + + if (avail > 0) + { + int used = creditManager.creditsReleased((int)avail); + + long num = availableProducerCredits.addAndGet(-used); + + if (num < 0) + { + log.warn("Available credits has gone negative"); + + exceededAvailableCredits = true; + } + } + } + } + + public void addSize(final ServerMessage message, final boolean add) throws Exception + { + long size = message.getMemoryEstimate(); + + if (add) + { + checkReleaseProducerFlowControlCredits(size); + + addSize(size); + } + else + { + checkReleaseProducerFlowControlCredits(-size); + + addSize(-size); + } + } + + public void addSize(final MessageReference reference, final boolean add) throws Exception + { + long size = reference.getMemoryEstimate(); + + if (add) + { + checkReleaseProducerFlowControlCredits(size); + + addSize(size); + } + else + { + checkReleaseProducerFlowControlCredits(-size); + + addSize(-size); + } + } + + private void addSize(final long size) throws Exception + { + if (addressFullMessagePolicy != AddressFullMessagePolicy.PAGE) + { addAddressSize(size); + pagingManager.addSize(size); return; @@ -268,10 +374,12 @@ { throw new IllegalStateException("PagingStore(" + getStoreName() + ") not initialized"); } + + boolean full = isFull(); - if (dropMessagesWhenFull) + if (addressFullMessagePolicy == AddressFullMessagePolicy.DROP) { - if (isDrop()) + if (full) { if (!printedDropMessagesWarning) { @@ -288,6 +396,10 @@ return false; } } + else if (addressFullMessagePolicy == AddressFullMessagePolicy.BLOCK) + { + return false; + } // We need to ensure a read lock, as depage could change the paging state currentPageLock.readLock().lock(); @@ -637,6 +749,11 @@ return new PageImpl(this.storeName, storageManager, fileFactory, file, page); } + + public ServerProducerCreditManager getProducerCreditManager() + { + return creditManager; + } // TestSupportPageStore ------------------------------------------ @@ -827,7 +944,7 @@ } } - postOffice.route(message, new RoutingContextImpl(depageTransaction)); + postOffice.route(message, depageTransaction); } if (!running) @@ -963,7 +1080,7 @@ } // To be used on isDropMessagesWhenFull - private boolean isDrop() + private boolean isFull() { return getMaxSizeBytes() > 0 && getAddressSize() > getMaxSizeBytes(); } Modified: trunk/src/main/org/hornetq/core/paging/impl/TestSupportPageStore.java =================================================================== --- trunk/src/main/org/hornetq/core/paging/impl/TestSupportPageStore.java 2009-10-30 15:12:21 UTC (rev 8170) +++ trunk/src/main/org/hornetq/core/paging/impl/TestSupportPageStore.java 2009-10-30 17:41:19 UTC (rev 8171) @@ -35,4 +35,6 @@ Page depage() throws Exception; void forceAnotherPage() throws Exception; + + boolean isExceededAvailableCredits(); } Modified: trunk/src/main/org/hornetq/core/persistence/impl/journal/JournalLargeServerMessage.java =================================================================== --- trunk/src/main/org/hornetq/core/persistence/impl/journal/JournalLargeServerMessage.java 2009-10-30 15:12:21 UTC (rev 8170) +++ trunk/src/main/org/hornetq/core/persistence/impl/journal/JournalLargeServerMessage.java 2009-10-30 17:41:19 UTC (rev 8171) @@ -16,11 +16,14 @@ import static org.hornetq.utils.DataConstants.SIZE_INT; import java.nio.ByteBuffer; +import java.util.concurrent.atomic.AtomicInteger; import org.hornetq.core.journal.SequentialFile; import org.hornetq.core.logging.Logger; +import org.hornetq.core.paging.PagingStore; import org.hornetq.core.remoting.spi.HornetQBuffer; 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; @@ -91,7 +94,7 @@ { file.open(); } - + storageManager.addBytesToLargeMessage(file, this.getMessageID(), bytes); bodySize += bytes.length; @@ -161,13 +164,13 @@ @Override public synchronized int getEncodeSize() { - return getPropertiesEncodeSize(); + return getHeadersAndPropertiesEncodeSize(); } @Override public void encode(final HornetQBuffer buffer) { - encodeProperties(buffer); + encodeHeadersAndProperties(buffer); } @Override @@ -183,22 +186,34 @@ // File still null, this wasn't supposed to happen ever. log.warn(e.getMessage(), e); } - decodeProperties(buffer); + decodeHeadersAndProperties(buffer); } - @Override - public int decrementRefCount() + private final AtomicInteger delayDeletionCount = new AtomicInteger(0); + + public synchronized void incrementDelayDeletionCount() { - int currentRefCount = super.decrementRefCount(); + this.delayDeletionCount.incrementAndGet(); + } + + public synchronized void decrementDelayDeletionCount() throws Exception + { + int count = this.delayDeletionCount.decrementAndGet(); + + if (count == 0) + { + checkDelete(); + } + } - // 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 files - if (currentRefCount <= 0) + private void checkDelete() throws Exception + { + if (getRefCount() <= 0) { if (linkMessage != null) { // This file is linked to another message, deleting the reference where it belongs on this case - linkMessage.decrementRefCount(); + linkMessage.decrementDelayDeletionCount(); } else { @@ -217,7 +232,21 @@ } } } + } + @Override + public synchronized int decrementRefCount(PagingStore pagingStore, MessageReference reference) throws Exception + { + int currentRefCount = super.decrementRefCount(pagingStore, reference); + + // 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 + // files + if (delayDeletionCount.get() <= 0) + { + checkDelete(); + } + return currentRefCount; } @@ -233,7 +262,7 @@ releaseResources(); storageManager.deleteFile(file); } - + public boolean isFileExists() throws Exception { SequentialFile localfile = storageManager.createFileForLargeMessage(getMessageID(), isStored()); @@ -249,20 +278,18 @@ if (memoryEstimate == -1) { // The body won't be on memory (aways on-file), so we don't consider this for paging - memoryEstimate = getPropertiesEncodeSize() + SIZE_INT + getEncodeSize() + (16 + 4) * 2 + 1; + memoryEstimate = getHeadersAndPropertiesEncodeSize() + SIZE_INT + getEncodeSize() + (16 + 4) * 2 + 1; } return memoryEstimate; } - - + @Override public void setStored() throws Exception { super.setStored(); releaseResources(); - - + if (file != null && linkMessage == null) { storageManager.completeLargeMessage(this); @@ -287,11 +314,10 @@ @Override public synchronized ServerMessage copy(final long newID) throws Exception { - // Incrementing the reference counting to avoid deletion of the linkedMessage - incrementRefCount(); - + incrementDelayDeletionCount(); + long idToUse = messageID; - + if (linkMessage != null) { idToUse = linkMessage.getMessageID(); @@ -299,12 +325,14 @@ SequentialFile newfile = storageManager.createFileForLargeMessage(idToUse, isStored()); - JournalLargeServerMessage newMessage = new JournalLargeServerMessage(linkMessage == null ? this : (JournalLargeServerMessage)linkMessage, newfile, newID); + ServerMessage newMessage = new JournalLargeServerMessage(linkMessage == null ? this + : (JournalLargeServerMessage)linkMessage, + newfile, + newID); return newMessage; } - - + public SequentialFile getFile() { return file; @@ -341,13 +369,13 @@ } } - /* (non-Javadoc) - * @see org.hornetq.core.server.LargeServerMessage#getLinkedMessage() - */ - public LargeServerMessage getLinkedMessage() - { - return linkMessage; - } +// /* (non-Javadoc) +// * @see org.hornetq.core.server.LargeServerMessage#getLinkedMessage() +// */ +// public LargeServerMessage getLinkedMessage() +// { +// return linkMessage; +// } /* (non-Javadoc) * @see org.hornetq.core.server.LargeServerMessage#setLinkedMessage(org.hornetq.core.server.LargeServerMessage) Modified: trunk/src/main/org/hornetq/core/persistence/impl/journal/JournalStorageManager.java =================================================================== --- trunk/src/main/org/hornetq/core/persistence/impl/journal/JournalStorageManager.java 2009-10-30 15:12:21 UTC (rev 8170) +++ trunk/src/main/org/hornetq/core/persistence/impl/journal/JournalStorageManager.java 2009-10-30 17:41:19 UTC (rev 8171) @@ -423,7 +423,7 @@ HornetQBuffer headerBuffer = ChannelBuffers.wrappedBuffer(header); - largeMessage.decodeProperties(headerBuffer); + largeMessage.decodeHeadersAndProperties(headerBuffer); largeMessage.setMessageID(id); @@ -678,7 +678,7 @@ try { LargeServerMessage serverMessage = parseLargeMessage(messages, buff); - serverMessage.decrementRefCount(); + serverMessage.decrementDelayDeletionCount(); } catch (Exception e) { @@ -923,7 +923,7 @@ { log.debug("Large message: " + msg.getMessageID() + " didn't have any associated reference, file will be deleted"); - msg.decrementRefCount(); + msg.decrementDelayDeletionCount(); } } @@ -963,7 +963,7 @@ messages.put(originalMessageID, originalMessage); } - originalMessage.incrementRefCount(); + originalMessage.incrementDelayDeletionCount(); largeMessage.setLinkedMessage(originalMessage); } Modified: trunk/src/main/org/hornetq/core/persistence/impl/nullpm/NullStorageLargeServerMessage.java =================================================================== --- trunk/src/main/org/hornetq/core/persistence/impl/nullpm/NullStorageLargeServerMessage.java 2009-10-30 15:12:21 UTC (rev 8170) +++ trunk/src/main/org/hornetq/core/persistence/impl/nullpm/NullStorageLargeServerMessage.java 2009-10-30 17:41:19 UTC (rev 8171) @@ -135,6 +135,22 @@ return true; } + /* (non-Javadoc) + * @see org.hornetq.core.server.LargeServerMessage#decrementDelayDeletionCount() + */ + public void decrementDelayDeletionCount() throws Exception + { + + } + + /* (non-Javadoc) + * @see org.hornetq.core.server.LargeServerMessage#incrementDelayDeletionCount() + */ + public void incrementDelayDeletionCount() + { + + } + // Package protected --------------------------------------------- // Protected ----------------------------------------------------- Modified: trunk/src/main/org/hornetq/core/persistence/impl/nullpm/NullStorageManager.java =================================================================== --- trunk/src/main/org/hornetq/core/persistence/impl/nullpm/NullStorageManager.java 2009-10-30 15:12:21 UTC (rev 8170) +++ trunk/src/main/org/hornetq/core/persistence/impl/nullpm/NullStorageManager.java 2009-10-30 17:41:19 UTC (rev 8171) @@ -194,7 +194,7 @@ HornetQBuffer headerBuffer = ChannelBuffers.wrappedBuffer(header); - largeMessage.decodeProperties(headerBuffer); + largeMessage.decodeHeadersAndProperties(headerBuffer); largeMessage.setMessageID(id); Modified: trunk/src/main/org/hornetq/core/postoffice/Bindings.java =================================================================== --- trunk/src/main/org/hornetq/core/postoffice/Bindings.java 2009-10-30 15:12:21 UTC (rev 8170) +++ trunk/src/main/org/hornetq/core/postoffice/Bindings.java 2009-10-30 17:41:19 UTC (rev 8171) @@ -38,7 +38,7 @@ void setRouteWhenNoConsumers(boolean takePriorityIntoAccount); - void redistribute(ServerMessage message, Queue originatingQueue, RoutingContext context) throws Exception; + boolean redistribute(ServerMessage message, Queue originatingQueue, RoutingContext context) throws Exception; void route(ServerMessage message, RoutingContext context) throws Exception; } Modified: trunk/src/main/org/hornetq/core/postoffice/PostOffice.java =================================================================== --- trunk/src/main/org/hornetq/core/postoffice/PostOffice.java 2009-10-30 15:12:21 UTC (rev 8170) +++ trunk/src/main/org/hornetq/core/postoffice/PostOffice.java 2009-10-30 17:41:19 UTC (rev 8171) @@ -54,11 +54,11 @@ void route(ServerMessage message) throws Exception; - void route(ServerMessage message, RoutingContext context) throws Exception; + void route(ServerMessage message, Transaction tx) throws Exception; MessageReference reroute(ServerMessage message, Queue queue, Transaction tx) throws Exception; - boolean redistribute(ServerMessage message, final Queue originatingQueue, RoutingContext context) throws Exception; + boolean redistribute(ServerMessage message, final Queue originatingQueue, Transaction tx) throws Exception; PagingManager getPagingManager(); Modified: trunk/src/main/org/hornetq/core/postoffice/impl/BindingsImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/postoffice/impl/BindingsImpl.java 2009-10-30 15:12:21 UTC (rev 8170) +++ trunk/src/main/org/hornetq/core/postoffice/impl/BindingsImpl.java 2009-10-30 17:41:19 UTC (rev 8171) @@ -15,28 +15,24 @@ import java.nio.ByteBuffer; import java.util.Collection; -import java.util.HashSet; 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.CopyOnWriteArrayList; +import org.hornetq.core.exception.HornetQException; import org.hornetq.core.filter.Filter; import org.hornetq.core.logging.Logger; import org.hornetq.core.message.impl.MessageImpl; import org.hornetq.core.postoffice.Binding; import org.hornetq.core.postoffice.Bindings; -import org.hornetq.core.server.Bindable; import org.hornetq.core.server.Queue; import org.hornetq.core.server.RoutingContext; import org.hornetq.core.server.ServerMessage; +import org.hornetq.core.server.group.GroupingHandler; import org.hornetq.core.server.group.impl.Proposal; import org.hornetq.core.server.group.impl.Response; -import org.hornetq.core.server.group.GroupingHandler; -import org.hornetq.core.transaction.Transaction; -import org.hornetq.core.exception.HornetQException; import org.hornetq.utils.SimpleString; /** @@ -135,86 +131,93 @@ bindingsMap.remove(binding.getID()); } - public void redistribute(final ServerMessage message, final Queue originatingQueue, final RoutingContext context) throws Exception + public boolean redistribute(final ServerMessage message, final Queue originatingQueue, final RoutingContext context) throws Exception + { + + if (routeWhenNoConsumers) { - if (routeWhenNoConsumers) - { - return; - } + return false; + } - SimpleString routingName = originatingQueue.getName(); + SimpleString routingName = originatingQueue.getName(); - List<Binding> bindings = routingNameBindingMap.get(routingName); + List<Binding> bindings = routingNameBindingMap.get(routingName); - if (bindings == null) - { - // The value can become null if it's concurrently removed while we're iterating - this is expected - // ConcurrentHashMap behaviour! - return; - } + if (bindings == null) + { + // The value can become null if it's concurrently removed while we're iterating - this is expected + // ConcurrentHashMap behaviour! + return false; + } - Integer ipos = routingNamePositions.get(routingName); + Integer ipos = routingNamePositions.get(routingName); - int pos = ipos != null ? ipos.intValue() : 0; + int pos = ipos != null ? ipos.intValue() : 0; - int length = bindings.size(); + int length = bindings.size(); - int startPos = pos; + int startPos = pos; - Binding theBinding = null; + Binding theBinding = null; - // TODO - combine this with similar logic in route() - while (true) + // TODO - combine this with similar logic in route() + while (true) + { + Binding binding; + try { - Binding binding; - try + binding = bindings.get(pos); + } + catch (IndexOutOfBoundsException e) + { + // This can occur if binding is removed while in route + if (!bindings.isEmpty()) { - binding = bindings.get(pos); + pos = 0; + startPos = 0; + length = bindings.size(); + + continue; } - catch (IndexOutOfBoundsException e) + else { - // This can occur if binding is removed while in route - if (!bindings.isEmpty()) - { - pos = 0; - startPos = 0; - length = bindings.size(); - - continue; - } - else - { - break; - } + break; } + } - pos = incrementPos(pos, length); + pos = incrementPos(pos, length); - Filter filter = binding.getFilter(); + Filter filter = binding.getFilter(); - boolean highPrior = binding.isHighAcceptPriority(message); + boolean highPrior = binding.isHighAcceptPriority(message); - if (highPrior && binding.getBindable() != originatingQueue && (filter == null || filter.match(message))) - { - theBinding = binding; + if (highPrior && binding.getBindable() != originatingQueue && (filter == null || filter.match(message))) + { + theBinding = binding; - break; - } - - if (pos == startPos) - { - break; - } + break; } - routingNamePositions.put(routingName, pos); - - if (theBinding != null) + if (pos == startPos) { - theBinding.route(message, context); + break; } } + routingNamePositions.put(routingName, pos); + + if (theBinding != null) + { + theBinding.route(message, context); + + return true; + } + else + { + return false; + } + } + public void route(final ServerMessage message, final RoutingContext context) throws Exception { boolean routed = false; @@ -259,7 +262,6 @@ Binding theBinding = getNextBinding(message, routingName, bindings); - if (theBinding != null) { theBinding.route(message, context); @@ -368,10 +370,11 @@ return theBinding; } - private void routeUsingStrictOrdering(ServerMessage message, RoutingContext context, GroupingHandler groupingGroupingHandler) - throws Exception + private void routeUsingStrictOrdering(final ServerMessage message, + final RoutingContext context, + final GroupingHandler groupingGroupingHandler) throws Exception { - SimpleString groupId = (SimpleString) message.getProperty(MessageImpl.HDR_GROUP_ID); + SimpleString groupId = (SimpleString)message.getProperty(MessageImpl.HDR_GROUP_ID); for (Map.Entry<SimpleString, List<Binding>> entry : routingNameBindingMap.entrySet()) { @@ -386,20 +389,20 @@ continue; } - //concat a full group id, this is for when a binding has multiple bindings + // concat a full group id, this is for when a binding has multiple bindings SimpleString fullID = groupId.concat(".").concat(routingName); - //see if there is already a response + // see if there is already a response Response resp = groupingGroupingHandler.getProposal(fullID); if (resp == null) { - //ok lets find the next binding to propose + // ok lets find the next binding to propose Binding theBinding = getNextBinding(message, routingName, bindings); - //TODO https://jira.jboss.org/jira/browse/HORNETQ-191 + // TODO https://jira.jboss.org/jira/browse/HORNETQ-191 resp = groupingGroupingHandler.propose(new Proposal(fullID, theBinding.getClusterName())); - //if our proposal was declined find the correct binding to use + // if our proposal was declined find the correct binding to use if (resp.getAlternativeClusterName() != null) { theBinding = null; @@ -413,19 +416,21 @@ } } - //and lets route it + // and lets route it if (theBinding != null) { theBinding.route(message, context); } else { - throw new HornetQException(HornetQException.QUEUE_DOES_NOT_EXIST, "queue " + resp.getChosenClusterName() + " has been removed cannot deliver message, queues should not be removed when grouping is used"); + throw new HornetQException(HornetQException.QUEUE_DOES_NOT_EXIST, + "queue " + resp.getChosenClusterName() + + " has been removed cannot deliver message, queues should not be removed when grouping is used"); } } else { - //ok, we need to find the binding and route it + // ok, we need to find the binding and route it Binding chosen = null; for (Binding binding : bindings) { @@ -441,14 +446,12 @@ } else { - throw new HornetQException(HornetQException.QUEUE_DOES_NOT_EXIST, "queue " + resp.getChosenClusterName() + " has been removed cannot deliver message, queues should not be removed when grouping is used"); + throw new HornetQException(HornetQException.QUEUE_DOES_NOT_EXIST, + "queue " + resp.getChosenClusterName() + + " has been removed cannot deliver message, queues should not be removed when grouping is used"); } } - - } - - } private void routeFromCluster(final ServerMessage message, final RoutingContext context) throws Exception Modified: trunk/src/main/org/hornetq/core/postoffice/impl/PostOfficeImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/postoffice/impl/PostOfficeImpl.java 2009-10-30 15:12:21 UTC (rev 8170) +++ trunk/src/main/org/hornetq/core/postoffice/impl/PostOfficeImpl.java 2009-10-30 17:41:19 UTC (rev 8171) @@ -20,7 +20,6 @@ import java.util.HashSet; import java.util.List; import java.util.Map; -import java.util.Set; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; @@ -43,12 +42,12 @@ import org.hornetq.core.postoffice.Binding; import org.hornetq.core.postoffice.BindingType; import org.hornetq.core.postoffice.Bindings; +import org.hornetq.core.postoffice.BindingsFactory; import org.hornetq.core.postoffice.DuplicateIDCache; import org.hornetq.core.postoffice.PostOffice; import org.hornetq.core.postoffice.QueueInfo; -import org.hornetq.core.postoffice.BindingsFactory; -import org.hornetq.core.server.MessageReference; import org.hornetq.core.server.HornetQServer; +import org.hornetq.core.server.MessageReference; import org.hornetq.core.server.Queue; import org.hornetq.core.server.QueueFactory; import org.hornetq.core.server.RoutingContext; @@ -434,7 +433,7 @@ public synchronized void addBinding(final Binding binding) throws Exception { addressManager.addBinding(binding); - + TypedProperties props = new TypedProperties(); props.putIntProperty(ManagementHelper.HDR_BINDING_TYPE, binding.getType().toInt()); @@ -528,13 +527,21 @@ public void route(final ServerMessage message) throws Exception { - route(message, new RoutingContextImpl(null)); + route(message, null); } - public void route(final ServerMessage message, final RoutingContext context) throws Exception + public void route(final ServerMessage message, final Transaction tx) throws Exception { - SimpleString address = message.getDestination(); + // Sanity check + if (message.getRefCount() > 0) + { + throw new IllegalStateException("Message cannot be routed more than once"); + } + RoutingContext context = new RoutingContextImpl(tx); + + SimpleString address = message.getDestination(); + Object duplicateID = message.getProperty(MessageImpl.HDR_DUPLICATE_DETECTION_ID); DuplicateIDCache cache = null; @@ -579,9 +586,9 @@ { // We need to store the duplicate id atomically with the message storage, so we need to create a tx for this - Transaction tx = new TransactionImpl(storageManager); + Transaction newTX = new TransactionImpl(storageManager); - context.setTransaction(tx); + context.setTransaction(newTX); startedTx = true; } @@ -616,58 +623,49 @@ if (bindings != null) { - context.incrementDepth(); - bindings.route(message, context); - - context.decrementDepth(); } - //The depth allows for recursion e.g. with diverts - we only want to process the route after any recursed routes - //have been processed - - if (context.getDepth() == 0) + if (context.getQueues().isEmpty()) { - if (context.getQueues().isEmpty()) - { - // Send to DLA if appropriate + // Send to DLA if appropriate AddressSettings addressSettings = addressSettingsRepository.getMatch(address.toString()); - + boolean sendToDLA = addressSettings.isSendToDLAOnNoRoute(); - + if (sendToDLA) { - //Send to the DLA for the address - + // Send to the DLA for the address + SimpleString dlaAddress = addressSettings.getDeadLetterAddress(); - + if (dlaAddress == null) { - log.warn("Did not route to any bindings for address " + address + " and sendToDLAOnNoRoute is true " + + log.warn("Did not route to any bindings for address " + address + + " and sendToDLAOnNoRoute is true " + "but there is no DLA configured for the address, the message will be ignored."); } else { message.setOriginalHeaders(message, false); - + message.setDestination(dlaAddress); - - route(message, context); + + route(message, tx); } } } - else - { - processRoute(message, context); - } + else + { + processRoute(message, context); + } if (startedTx) { - context.getTransaction().commit(); + context.getTransaction().commit(); } } - } public MessageReference reroute(final ServerMessage message, final Queue queue, final Transaction tx) throws Exception { @@ -678,23 +676,16 @@ if (scheduledDeliveryTime != null) { reference.setScheduledDeliveryTime(scheduledDeliveryTime); - } + } message.incrementDurableRefCount(); message.setStored(); - int refCount = message.incrementRefCount(); - PagingStore store = pagingManager.getPageStore(message.getDestination()); - if (refCount == 1) - { - store.addSize(message.getMemoryEstimate()); - } + message.incrementRefCount(store, reference); - store.addSize(reference.getMemoryEstimate()); - if (tx == null) { queue.addLast(reference); @@ -711,7 +702,7 @@ return reference; } - public boolean redistribute(final ServerMessage message, final Queue originatingQueue, final RoutingContext context) throws Exception + public boolean redistribute(final ServerMessage message, final Queue originatingQueue, final Transaction tx) throws Exception { Bindings bindings = addressManager.getBindingsForRoutingAddress(message.getDestination()); @@ -719,9 +710,11 @@ if (bindings != null) { - bindings.redistribute(message, originatingQueue, context); + RoutingContext context = new RoutingContextImpl(tx); - if (!context.getQueues().isEmpty()) + boolean routed = bindings.redistribute(message, originatingQueue, context); + + if (routed) { processRoute(message, context); @@ -853,11 +846,13 @@ } private void processRoute(final ServerMessage message, final RoutingContext context) throws Exception - { + { final List<MessageReference> refs = new ArrayList<MessageReference>(); Transaction tx = context.getTransaction(); + PagingStore store = pagingManager.getPageStore(message.getDestination()); + for (Queue queue : context.getQueues()) { MessageReference reference = message.createReference(queue); @@ -869,7 +864,7 @@ if (scheduledDeliveryTime != null) { reference.setScheduledDeliveryTime(scheduledDeliveryTime); - } + } if (message.isDurable() && queue.isDurable()) { @@ -880,7 +875,7 @@ if (tx != null) { storageManager.storeMessageTransactional(tx.getID(), message); - } + } else { storageManager.storeMessage(message); @@ -890,22 +885,22 @@ } if (tx != null) - { + { storageManager.storeReferenceTransactional(tx.getID(), queue.getID(), message.getMessageID()); tx.putProperty(TransactionPropertyIndexes.CONTAINS_PERSISTENT, true); } else - { + { storageManager.storeReference(queue.getID(), message.getMessageID()); - } + } if (scheduledDeliveryTime != null) { if (tx != null) { storageManager.updateScheduledDeliveryTimeTransactional(tx.getID(), reference); - } + } else { storageManager.updateScheduledDeliveryTime(reference); @@ -913,16 +908,7 @@ } } - int refCount = message.incrementRefCount(); - - PagingStore store = pagingManager.getPageStore(message.getDestination()); - - if (refCount == 1) - { - store.addSize(message.getMemoryEstimate()); - } - - store.addSize(reference.getMemoryEstimate()); + message.incrementRefCount(store, reference); } if (tx != null) @@ -1217,14 +1203,14 @@ AddOperation(final List<MessageReference> refs) { this.refs = refs; - } + } public void afterCommit(Transaction tx) { for (MessageReference ref : refs) { ref.getQueue().addLast(ref); -} + } } public void afterPrepare(Transaction tx) throws Exception @@ -1256,16 +1242,9 @@ message.decrementDurableRefCount(); } - int count = message.decrementRefCount(); - PagingStore store = pagingManager.getPageStore(message.getDestination()); - if (count == 0) - { - store.addSize(-message.getMemoryEstimate()); - } - - store.addSize(-ref.getMemoryEstimate()); + message.decrementRefCount(store, ref); } } } Modified: trunk/src/main/org/hornetq/core/remoting/Channel.java =================================================================== --- trunk/src/main/org/hornetq/core/remoting/Channel.java 2009-10-30 15:12:21 UTC (rev 8170) +++ trunk/src/main/org/hornetq/core/remoting/Channel.java 2009-10-30 17:41:19 UTC (rev 8171) @@ -62,4 +62,6 @@ void handlePacket(Packet packet); void clearCommands(); + + int getConfirmationWindowSize(); } Modified: trunk/src/main/org/hornetq/core/remoting/FailureListener.java =================================================================== --- trunk/src/main/org/hornetq/core/remoting/FailureListener.java 2009-10-30 15:12:21 UTC (rev 8170) +++ trunk/src/main/org/hornetq/core/remoting/FailureListener.java 2009-10-30 17:41:19 UTC (rev 8171) @@ -25,8 +25,5 @@ */ public interface FailureListener { - /** - * @return <code>false</code> to stop calling subsequent failure listeners - */ void connectionFailed(HornetQException me); } Modified: trunk/src/main/org/hornetq/core/remoting/RemotingConnection.java =================================================================== --- trunk/src/main/org/hornetq/core/remoting/RemotingConnection.java 2009-10-30 15:12:21 UTC (rev 8170) +++ trunk/src/main/org/hornetq/core/remoting/RemotingConnection.java 2009-10-30 17:41:19 UTC (rev 8171) @@ -32,7 +32,7 @@ String getRemoteAddress(); - Channel getChannel(long channelID, int windowSize, boolean block); + Channel getChannel(long channelID, int confWindowSize); void putChannel(long channelID, Channel channel); Modified: trunk/src/main/org/hornetq/core/remoting/impl/ChannelImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/remoting/impl/ChannelImpl.java 2009-10-30 15:12:21 UTC (rev 8170) +++ trunk/src/main/org/hornetq/core/remoting/impl/ChannelImpl.java 2009-10-30 17:41:19 UTC (rev 8171) @@ -16,7 +16,6 @@ import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.PACKETS_CONFIRMED; import java.util.concurrent.ConcurrentLinkedQueue; -import java.util.concurrent.Semaphore; import java.util.concurrent.TimeUnit; import java.util.concurrent.locks.Condition; import java.util.concurrent.locks.Lock; @@ -71,44 +70,27 @@ private boolean failingOver; - private final int windowSize; - private final int confWindowSize; - private volatile Semaphore sendSemaphore; - private int receivedBytes; private CommandConfirmationHandler commandConfirmationHandler; - public ChannelImpl(final RemotingConnection connection, final long id, final int windowSize, final boolean block) + public ChannelImpl(final RemotingConnection connection, final long id, final int confWindowSize) { this.connection = connection; this.id = id; - this.windowSize = windowSize; + this.confWindowSize = confWindowSize; - this.confWindowSize = (int)(0.75 * windowSize); - - if (this.windowSize != -1) + if (confWindowSize != -1) { resendCache = new ConcurrentLinkedQueue<Packet>(); - - if (block) - { - sendSemaphore = new Semaphore(windowSize, true); - } - else - { - sendSemaphore = null; - } } else { resendCache = null; - - sendSemaphore = null; } } @@ -126,6 +108,11 @@ { return lock; } + + public int getConfirmationWindowSize() + { + return confWindowSize; + } public void returnBlocking() { @@ -158,14 +145,14 @@ // This must never called by more than one thread concurrently public void send(final Packet packet, final boolean flush) - { + { synchronized (sendLock) { packet.setChannelID(id); final HornetQBuffer buffer = connection.getTransportConnection().createBuffer(packet.getRequiredBufferSize()); - int size = packet.encode(buffer); + packet.encode(buffer); lock.lock(); @@ -194,20 +181,6 @@ { lock.unlock(); } - - // Must block on semaphore outside the main lock or this can prevent failover from occurring, also after the - // packet is sent to assure we get some credits back - if (sendSemaphore != null && packet.getType() != PACKETS_CONFIRMED) - { - try - { - sendSemaphore.acquire(size); - } - catch (InterruptedException e) - { - throw new IllegalStateException("Semaphore interrupted"); - } - } } } @@ -231,7 +204,7 @@ final HornetQBuffer buffer = connection.getTransportConnection().createBuffer(packet.getRequiredBufferSize()); - int size = packet.encode(buffer); + packet.encode(buffer); lock.lock(); @@ -293,9 +266,9 @@ if (response.getType() == PacketImpl.EXCEPTION) { final HornetQExceptionMessage mem = (HornetQExceptionMessage)response; - + HornetQException e = mem.getException(); - + e.fillInStackTrace(); throw e; @@ -305,19 +278,7 @@ { lock.unlock(); } - // Must block on semaphore outside the main lock or this can prevent failover from occurring, also after the - // packet is sent to assure we get some credits back - if (sendSemaphore != null && packet.getType() != PACKETS_CONFIRMED) - { - try - { - sendSemaphore.acquire(size); - } - catch (InterruptedException e) - { - throw new IllegalStateException("Semaphore interrupted"); - } - } + return response; } } @@ -343,14 +304,6 @@ { return; } - - if (sendSemaphore != null) - { - //Any threads blocking on the send semaphore should be allowed to return - we do this by just giving it - //a lot of permits - note we don't give it Integer.MAX_VALUE since then if if more releases come in that - //could end up with permit count going -ve which would cause subsequent sends to block - sendSemaphore.release(Integer.MAX_VALUE / 2); - } if (!connection.isDestroyed() && !connection.removeChannel(id)) { @@ -380,13 +333,16 @@ public void replayCommands(final int otherLastReceivedCommandID, final long newChannelID) { - clearUpTo(otherLastReceivedCommandID); - - for (final Packet packet : resendCache) + if (resendCache != null) { - packet.setChannelID(newChannelID); - - doWrite(packet); + clearUpTo(otherLastReceivedCommandID); + + for (final Packet packet : resendCache) + { + packet.setChannelID(newChannelID); + + doWrite(packet); + } } } @@ -420,7 +376,7 @@ if (receivedBytes != 0) { receivedBytes = 0; - + final Packet confirmed = new PacketsConfirmedMessage(lastReceivedCommandID); confirmed.setChannelID(id); @@ -441,7 +397,6 @@ { receivedBytes = 0; - final Packet confirmed = new PacketsConfirmedMessage(lastReceivedCommandID); confirmed.setChannelID(id); @@ -450,26 +405,16 @@ } } } - + public void clearCommands() { - lastReceivedCommandID = -1; - - firstStoredCommandID = 0; - - resendCache.clear(); - - Semaphore oldSemaphore = sendSemaphore; - - if (oldSemaphore != null) + if (resendCache != null) { - //Reset the semaphore - sendSemaphore = new Semaphore(windowSize, true); - - //Any threads blocking on the send semaphore should be allowed to return - we do this by just giving it - //a lot of permits - note we don't give it Integer.MAX_VALUE since then if if more releases come in that - //could end up with permit count going -ve which would cause subsequent sends to block - oldSemaphore.release(Integer.MAX_VALUE / 2); + lastReceivedCommandID = -1; + + firstStoredCommandID = 0; + + resendCache.clear(); } } @@ -543,11 +488,10 @@ if (packet == null) { - log.warn("Can't find packet to clear: " + - " last received command id " + - lastReceivedCommandID + - " first stored command id " + - firstStoredCommandID); + log.warn("Can't find packet to clear: " + " last received command id " + + lastReceivedCommandID + + " first stored command id " + + firstStoredCommandID); return; } @@ -563,10 +507,5 @@ } firstStoredCommandID += numberToClear; - - if (sendSemaphore != null) - { - sendSemaphore.release(sizeToFree); - } } } Modified: trunk/src/main/org/hornetq/core/remoting/impl/PacketDecoder.java =================================================================== --- trunk/src/main/org/hornetq/core/remoting/impl/PacketDecoder.java 2009-10-30 15:12:21 UTC (rev 8170) +++ trunk/src/main/org/hornetq/core/remoting/impl/PacketDecoder.java 2009-10-30 17:41:19 UTC (rev 8171) @@ -13,22 +13,10 @@ package org.hornetq.core.remoting.impl; -import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.REPLICATION_LARGE_MESSAGE_BEGIN; -import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.REPLICATION_LARGE_MESSAGE_END; -import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.REPLICATION_LARGE_MESSAGE_WRITE; -import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.REPLICATION_PAGE_EVENT; -import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.REPLICATION_PAGE_WRITE; -import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.REPLICATION_PREPARE; -import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.REPLICATION_DELETE_TX; -import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.REPLICATION_COMMIT_ROLLBACK; -import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.REPLICATION_APPEND_TX; -import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.REPLICATION_DELETE; -import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.REPLICATION_RESPONSE; -import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.REPLICATION_APPEND; -import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.CREATE_REPLICATION; import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.CREATESESSION; import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.CREATESESSION_RESP; import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.CREATE_QUEUE; +import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.CREATE_REPLICATION; import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.DELETE_QUEUE; import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.DISCONNECT; import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.EXCEPTION; @@ -37,16 +25,30 @@ import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.PING; import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.REATTACH_SESSION; import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.REATTACH_SESSION_RESP; +import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.REPLICATION_APPEND; +import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.REPLICATION_APPEND_TX; +import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.REPLICATION_COMMIT_ROLLBACK; +import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.REPLICATION_DELETE; +import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.REPLICATION_DELETE_TX; +import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.REPLICATION_LARGE_MESSAGE_BEGIN; +import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.REPLICATION_LARGE_MESSAGE_END; +import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.REPLICATION_LARGE_MESSAGE_WRITE; +import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.REPLICATION_PAGE_EVENT; +import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.REPLICATION_PAGE_WRITE; +import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.REPLICATION_PREPARE; +import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.REPLICATION_RESPONSE; import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.SESS_ACKNOWLEDGE; import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.SESS_BINDINGQUERY; import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.SESS_BINDINGQUERY_RESP; import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.SESS_CLOSE; import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.SESS_COMMIT; import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.SESS_CONSUMER_CLOSE; -import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.SESS_FORCE_CONSUMER_DELIVERY; import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.SESS_CREATECONSUMER; import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.SESS_EXPIRED; import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.SESS_FLOWTOKEN; +import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.SESS_FORCE_CONSUMER_DELIVERY; +import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.SESS_PRODUCER_CREDITS; +import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.SESS_PRODUCER_REQUEST_CREDITS; import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.SESS_QUEUEQUERY; import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.SESS_QUEUEQUERY_RESP; import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.SESS_RECEIVE_CONTINUATION; @@ -104,15 +106,17 @@ import org.hornetq.core.remoting.impl.wireformat.SessionBindingQueryResponseMessage; import org.hornetq.core.remoting.impl.wireformat.SessionCloseMessage; import org.hornetq.core.remoting.impl.wireformat.SessionConsumerCloseMessage; -import org.hornetq.core.remoting.impl.wireformat.SessionForceConsumerDelivery; import org.hornetq.core.remoting.impl.wireformat.SessionConsumerFlowCreditMessage; import org.hornetq.core.remoting.impl.wireformat.SessionCreateConsumerMessage; import org.hornetq.core.remoting.impl.wireformat.SessionDeleteQueueMessage; import org.hornetq.core.remoting.impl.wireformat.SessionExpiredMessage; +import org.hornetq.core.remoting.impl.wireformat.SessionForceConsumerDelivery; +import org.hornetq.core.remoting.impl.wireformat.SessionProducerCreditsMessage; import org.hornetq.core.remoting.impl.wireformat.SessionQueueQueryMessage; import org.hornetq.core.remoting.impl.wireformat.SessionQueueQueryResponseMessage; import org.hornetq.core.remoting.impl.wireformat.SessionReceiveContinuationMessage; import org.hornetq.core.remoting.impl.wireformat.SessionReceiveMessage; +import org.hornetq.core.remoting.impl.wireformat.SessionRequestProducerCreditsMessage; import org.hornetq.core.remoting.impl.wireformat.SessionSendContinuationMessage; import org.hornetq.core.remoting.impl.wireformat.SessionSendLargeMessage; import org.hornetq.core.remoting.impl.wireformat.SessionSendMessage; @@ -378,6 +382,16 @@ packet = new SessionSendContinuationMessage(); break; } + case SESS_PRODUCER_REQUEST_CREDITS: + { + packet = new SessionRequestProducerCreditsMessage(); + break; + } + case SESS_PRODUCER_CREDITS: + { + packet = new SessionProducerCreditsMessage(); + break; + } case CREATE_REPLICATION: { packet = new CreateReplicationSessionMessage(); Modified: trunk/src/main/org/hornetq/core/remoting/impl/RemotingConnectionImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/remoting/impl/RemotingConnectionImpl.java 2009-10-30 15:12:21 UTC (rev 8170) +++ trunk/src/main/org/hornetq/core/remoting/impl/RemotingConnectionImpl.java 2009-10-30 17:41:19 UTC (rev 8171) @@ -20,7 +20,6 @@ import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.Executor; -import org.hornetq.core.client.ClientMessage; import org.hornetq.core.exception.HornetQException; import org.hornetq.core.logging.Logger; import org.hornetq.core.remoting.Channel; @@ -30,11 +29,9 @@ import org.hornetq.core.remoting.Packet; import org.hornetq.core.remoting.RemotingConnection; import org.hornetq.core.remoting.impl.wireformat.PacketImpl; -import org.hornetq.core.remoting.impl.wireformat.SessionReceiveMessage; import org.hornetq.core.remoting.spi.Connection; import org.hornetq.core.remoting.spi.HornetQBuffer; import org.hornetq.utils.SimpleIDGenerator; -import org.hornetq.utils.SimpleString; /** * @author <a href=&quot;tim.fox(a)jboss.com&quot;&gt;Tim Fox</a> @@ -80,8 +77,6 @@ private final Object transferLock = new Object(); - // private boolean frozen; - private final Object failLock = new Object(); private final PacketDecoder decoder = new PacketDecoder(); @@ -162,13 +157,13 @@ return transportConnection.getRemoteAddress(); } - public synchronized Channel getChannel(final long channelID, final int windowSize, final boolean block) + public synchronized Channel getChannel(final long channelID, final int confWindowSize) { Channel channel = channels.get(channelID); if (channel == null) { - channel = new ChannelImpl(this, channelID, windowSize, block); + channel = new ChannelImpl(this, channelID, confWindowSize); channels.put(channelID, channel); } @@ -363,7 +358,7 @@ try { boolean callNext = interceptor.intercept(packet, this); - + if (!callNext) { // abort Modified: trunk/src/main/org/hornetq/core/remoting/impl/wireformat/PacketImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/remoting/impl/wireformat/PacketImpl.java 2009-10-30 15:12:21 UTC (rev 8170) +++ trunk/src/main/org/hornetq/core/remoting/impl/wireformat/PacketImpl.java 2009-10-30 17:41:19 UTC (rev 8171) @@ -140,6 +140,10 @@ public static final byte SESS_RECEIVE_CONTINUATION = 76; public static final byte SESS_FORCE_CONSUMER_DELIVERY = 77; + + public static final byte SESS_PRODUCER_REQUEST_CREDITS = 78; + + public static final byte SESS_PRODUCER_CREDITS = 79; // Replication @@ -212,7 +216,7 @@ public void decode(final HornetQBuffer buffer) { channelID = buffer.readLong(); - + decodeBody(buffer); size = buffer.readerIndex(); Added: trunk/src/main/org/hornetq/core/remoting/impl/wireformat/SessionProducerCreditsMessage.java =================================================================== --- trunk/src/main/org/hornetq/core/remoting/impl/wireformat/SessionProducerCreditsMessage.java (rev 0) +++ trunk/src/main/org/hornetq/core/remoting/impl/wireformat/SessionProducerCreditsMessage.java 2009-10-30 17:41:19 UTC (rev 8171) @@ -0,0 +1,97 @@ +/* + * 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.remoting.impl.wireformat; + +import org.hornetq.core.remoting.spi.HornetQBuffer; +import org.hornetq.utils.DataConstants; +import org.hornetq.utils.SimpleString; + +/** + * @author <a href="mailto:tim.fox@jboss.com">Tim Fox</a> + * + */ +public class SessionProducerCreditsMessage extends PacketImpl +{ + // Constants ----------------------------------------------------- + + // Attributes ---------------------------------------------------- + + private int credits; + + private SimpleString address; + + // Static -------------------------------------------------------- + + // Constructors -------------------------------------------------- + + public SessionProducerCreditsMessage(final int credits, final SimpleString address) + { + super(SESS_PRODUCER_CREDITS); + + this.credits = credits; + + this.address = address; + } + + public SessionProducerCreditsMessage() + { + super(SESS_PRODUCER_CREDITS); + } + + // Public -------------------------------------------------------- + + public int getCredits() + { + return credits; + } + + public SimpleString getAddress() + { + return address; + } + +// public boolean isRequiresConfirmations() +// { +// return false; +// } + + @Override + public void encodeBody(final HornetQBuffer buffer) + { + buffer.writeInt(credits); + buffer.writeSimpleString(address); + } + + @Override + public void decodeBody(final HornetQBuffer buffer) + { + credits = buffer.readInt(); + address = buffer.readSimpleString(); + } + + public int getRequiredBufferSize() + { + int size = BASIC_PACKET_SIZE + DataConstants.SIZE_INT + SimpleString.sizeofString(address); + + return size; + } + + // Package protected --------------------------------------------- + + // Protected ----------------------------------------------------- + + // Private ------------------------------------------------------- + + // Inner classes ------------------------------------------------- +} Added: trunk/src/main/org/hornetq/core/remoting/impl/wireformat/SessionRequestProducerCreditsMessage.java =================================================================== --- trunk/src/main/org/hornetq/core/remoting/impl/wireformat/SessionRequestProducerCreditsMessage.java (rev 0) +++ trunk/src/main/org/hornetq/core/remoting/impl/wireformat/SessionRequestProducerCreditsMessage.java 2009-10-30 17:41:19 UTC (rev 8171) @@ -0,0 +1,97 @@ +/* + * 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.remoting.impl.wireformat; + +import org.hornetq.core.remoting.spi.HornetQBuffer; +import org.hornetq.utils.DataConstants; +import org.hornetq.utils.SimpleString; + +/** + * @author <a href="mailto:tim.fox@jboss.com">Tim Fox</a> + * + */ +public class SessionRequestProducerCreditsMessage extends PacketImpl +{ + // Constants ----------------------------------------------------- + + // Attributes ---------------------------------------------------- + + private int credits; + + private SimpleString address; + + // Static -------------------------------------------------------- + + // Constructors -------------------------------------------------- + + public SessionRequestProducerCreditsMessage(final int credits, final SimpleString address) + { + super(SESS_PRODUCER_REQUEST_CREDITS); + + this.credits = credits; + + this.address = address; + } + + public SessionRequestProducerCreditsMessage() + { + super(SESS_PRODUCER_REQUEST_CREDITS); + } + + // Public -------------------------------------------------------- + + public int getCredits() + { + return credits; + } + + public SimpleString getAddress() + { + return address; + } + +// public boolean isRequiresConfirmations() +// { +// return false; +// } + + @Override + public void encodeBody(final HornetQBuffer buffer) + { + buffer.writeInt(credits); + buffer.writeSimpleString(address); + } + + @Override + public void decodeBody(final HornetQBuffer buffer) + { + credits = buffer.readInt(); + address = buffer.readSimpleString(); + } + + public int getRequiredBufferSize() + { + int size = BASIC_PACKET_SIZE + DataConstants.SIZE_INT + SimpleString.sizeofString(address); + + return size; + } + + // Package protected --------------------------------------------- + + // Protected ----------------------------------------------------- + + // Private ------------------------------------------------------- + + // Inner classes ------------------------------------------------- +} Modified: trunk/src/main/org/hornetq/core/remoting/server/impl/RemotingServiceImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/remoting/server/impl/RemotingServiceImpl.java 2009-10-30 15:12:21 UTC (rev 8170) +++ trunk/src/main/org/hornetq/core/remoting/server/impl/RemotingServiceImpl.java 2009-10-30 17:41:19 UTC (rev 8171) @@ -230,7 +230,7 @@ for (ConnectionEntry entry : connections.values()) { - entry.connection.getChannel(0, -1, false).sendAndFlush(new PacketImpl(DISCONNECT)); + entry.connection.getChannel(0, -1).sendAndFlush(new PacketImpl(DISCONNECT)); } for (Acceptor acceptor : acceptors) @@ -301,25 +301,24 @@ .getExecutor() : null); - Channel channel1 = rc.getChannel(1, -1, false); + Channel channel1 = rc.getChannel(1, -1); ChannelHandler handler = createHandler(rc, channel1); channel1.setHandler(handler); - - - - + long ttl = ClientSessionFactoryImpl.DEFAULT_CONNECTION_TTL; + if (config.getConnectionTTLOverride() != -1) { ttl = config.getConnectionTTLOverride(); } + final ConnectionEntry entry = new ConnectionEntry(rc, System.currentTimeMillis(), ttl); connections.put(connection.getID(), entry); - final Channel channel0 = rc.getChannel(0, -1, false); + final Channel channel0 = rc.getChannel(0, -1); channel0.setHandler(new ChannelHandler() { Modified: trunk/src/main/org/hornetq/core/replication/impl/ReplicationManagerImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/replication/impl/ReplicationManagerImpl.java 2009-10-30 15:12:21 UTC (rev 8170) +++ trunk/src/main/org/hornetq/core/replication/impl/ReplicationManagerImpl.java 2009-10-30 17:41:19 UTC (rev 8171) @@ -18,6 +18,7 @@ import java.util.concurrent.ConcurrentLinkedQueue; import java.util.concurrent.Executor; +import org.hornetq.core.client.SessionFailureListener; import org.hornetq.core.client.impl.FailoverManager; import org.hornetq.core.exception.HornetQException; import org.hornetq.core.journal.EncodingSupport; @@ -25,7 +26,6 @@ import org.hornetq.core.paging.PagedMessage; import org.hornetq.core.remoting.Channel; import org.hornetq.core.remoting.ChannelHandler; -import org.hornetq.core.remoting.FailureListener; import org.hornetq.core.remoting.Packet; import org.hornetq.core.remoting.RemotingConnection; import org.hornetq.core.remoting.impl.wireformat.CreateReplicationSessionMessage; @@ -63,7 +63,7 @@ // Attributes ---------------------------------------------------- // TODO: where should this be configured? - private static final int WINDOW_SIZE = 1024 * 1024; + private static final int CONF_WINDOW_SIZE = 1024 * 1024; private final ResponseHandler responseHandler = new ResponseHandler(); @@ -319,18 +319,18 @@ long channelID = connection.generateChannelID(); - Channel mainChannel = connection.getChannel(1, -1, false); + Channel mainChannel = connection.getChannel(1, -1); - replicatingChannel = connection.getChannel(channelID, WINDOW_SIZE, false); + replicatingChannel = connection.getChannel(channelID, CONF_WINDOW_SIZE); replicatingChannel.setHandler(responseHandler); CreateReplicationSessionMessage replicationStartPackage = new CreateReplicationSessionMessage(channelID, - WINDOW_SIZE); + CONF_WINDOW_SIZE); mainChannel.sendBlocking(replicationStartPackage); - failoverManager.addFailureListener(new FailureListener() + failoverManager.addFailureListener(new SessionFailureListener() { public void connectionFailed(HornetQException me) { @@ -344,6 +344,10 @@ log.warn(e.getMessage(), e); } } + + public void beforeReconnect(HornetQException me) + { + } }); started = true; Modified: trunk/src/main/org/hornetq/core/server/HornetQServer.java =================================================================== --- trunk/src/main/org/hornetq/core/server/HornetQServer.java 2009-10-30 15:12:21 UTC (rev 8170) +++ trunk/src/main/org/hornetq/core/server/HornetQServer.java 2009-10-30 17:41:19 UTC (rev 8171) @@ -86,7 +86,7 @@ boolean autoCommitAcks, boolean preAcknowledge, boolean xa, - int producerWindowSize) throws Exception; + int confirmationWindowSize) throws Exception; void removeSession(String name) throws Exception; Modified: trunk/src/main/org/hornetq/core/server/LargeServerMessage.java =================================================================== --- trunk/src/main/org/hornetq/core/server/LargeServerMessage.java 2009-10-30 15:12:21 UTC (rev 8170) +++ trunk/src/main/org/hornetq/core/server/LargeServerMessage.java 2009-10-30 17:41:19 UTC (rev 8171) @@ -29,9 +29,6 @@ /** When a large message is copied (e.g. ExpiryQueue) instead of copying the file, we specify a link between the messages */ void setLinkedMessage(LargeServerMessage message); - /** When a large message is copied (e.g. ExpiryQueue) instead of copying the file, we specify a link between the messages */ - LargeServerMessage getLinkedMessage(); - boolean isFileExists() throws Exception; /** Close the files if opened */ @@ -40,4 +37,8 @@ long getLargeBodySize(); void deleteFile() throws Exception; + + void incrementDelayDeletionCount(); + + void decrementDelayDeletionCount() throws Exception; } Modified: trunk/src/main/org/hornetq/core/server/RoutingContext.java =================================================================== --- trunk/src/main/org/hornetq/core/server/RoutingContext.java 2009-10-30 15:12:21 UTC (rev 8170) +++ trunk/src/main/org/hornetq/core/server/RoutingContext.java 2009-10-30 17:41:19 UTC (rev 8171) @@ -34,9 +34,9 @@ List<Queue> getQueues(); - void incrementDepth(); + //void incrementDepth(); - void decrementDepth(); + // void decrementDepth(); - int getDepth(); + //int getDepth(); } Modified: trunk/src/main/org/hornetq/core/server/ServerMessage.java =================================================================== --- trunk/src/main/org/hornetq/core/server/ServerMessage.java 2009-10-30 15:12:21 UTC (rev 8170) +++ trunk/src/main/org/hornetq/core/server/ServerMessage.java 2009-10-30 17:41:19 UTC (rev 8171) @@ -15,6 +15,7 @@ import org.hornetq.core.journal.EncodingSupport; import org.hornetq.core.message.Message; +import org.hornetq.core.paging.PagingStore; /** * @@ -30,14 +31,16 @@ MessageReference createReference(Queue queue); - int incrementRefCount(); + int incrementRefCount(PagingStore pagingStore, MessageReference reference) + throws Exception; + int decrementRefCount(PagingStore pagingStore, MessageReference reference) + throws Exception; + int incrementDurableRefCount(); int decrementDurableRefCount(); - int decrementRefCount(); - ServerMessage copy(long newID) throws Exception; ServerMessage copy() throws Exception; Modified: trunk/src/main/org/hornetq/core/server/ServerSession.java =================================================================== --- trunk/src/main/org/hornetq/core/server/ServerSession.java 2009-10-30 15:12:21 UTC (rev 8170) +++ trunk/src/main/org/hornetq/core/server/ServerSession.java 2009-10-30 17:41:19 UTC (rev 8171) @@ -21,12 +21,13 @@ import org.hornetq.core.remoting.impl.wireformat.SessionAcknowledgeMessage; import org.hornetq.core.remoting.impl.wireformat.SessionBindingQueryMessage; import org.hornetq.core.remoting.impl.wireformat.SessionConsumerCloseMessage; -import org.hornetq.core.remoting.impl.wireformat.SessionForceConsumerDelivery; import org.hornetq.core.remoting.impl.wireformat.SessionConsumerFlowCreditMessage; import org.hornetq.core.remoting.impl.wireformat.SessionCreateConsumerMessage; import org.hornetq.core.remoting.impl.wireformat.SessionDeleteQueueMessage; import org.hornetq.core.remoting.impl.wireformat.SessionExpiredMessage; +import org.hornetq.core.remoting.impl.wireformat.SessionForceConsumerDelivery; import org.hornetq.core.remoting.impl.wireformat.SessionQueueQueryMessage; +import org.hornetq.core.remoting.impl.wireformat.SessionRequestProducerCreditsMessage; import org.hornetq.core.remoting.impl.wireformat.SessionSendContinuationMessage; import org.hornetq.core.remoting.impl.wireformat.SessionSendLargeMessage; import org.hornetq.core.remoting.impl.wireformat.SessionSendMessage; @@ -126,6 +127,9 @@ void handleSendLargeMessage(SessionSendLargeMessage packet); void handleForceConsumerDelivery(SessionForceConsumerDelivery message); + + void handleRequestProducerCredits(SessionRequestProducerCreditsMessage message) + throws Exception; void handleClose(Packet packet); @@ -136,5 +140,4 @@ ServerSessionPacketHandler getHandler(); void setHandler(ServerSessionPacketHandler handler); - } Modified: trunk/src/main/org/hornetq/core/server/cluster/impl/BridgeImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/server/cluster/impl/BridgeImpl.java 2009-10-30 15:12:21 UTC (rev 8170) +++ trunk/src/main/org/hornetq/core/server/cluster/impl/BridgeImpl.java 2009-10-30 17:41:19 UTC (rev 8171) @@ -25,6 +25,7 @@ import org.hornetq.core.client.ClientProducer; import org.hornetq.core.client.ClientSessionFactory; import org.hornetq.core.client.SendAcknowledgementHandler; +import org.hornetq.core.client.SessionFailureListener; import org.hornetq.core.client.impl.ClientSessionFactoryImpl; import org.hornetq.core.client.impl.ClientSessionInternal; import org.hornetq.core.client.management.impl.ManagementHelper; @@ -41,7 +42,6 @@ import org.hornetq.core.message.impl.MessageImpl; import org.hornetq.core.persistence.StorageManager; import org.hornetq.core.postoffice.BindingType; -import org.hornetq.core.remoting.FailureListener; import org.hornetq.core.remoting.RemotingConnection; import org.hornetq.core.server.HandleStatus; import org.hornetq.core.server.MessageReference; @@ -66,7 +66,7 @@ * * */ -public class BridgeImpl implements Bridge, FailureListener, SendAcknowledgementHandler +public class BridgeImpl implements Bridge, SessionFailureListener, SendAcknowledgementHandler { // Constants ----------------------------------------------------- @@ -241,7 +241,7 @@ } private void cancelRefs() throws Exception - { + { MessageReference ref; LinkedList<MessageReference> list = new LinkedList<MessageReference>(); @@ -257,11 +257,7 @@ queue = ref2.getQueue(); queue.cancel(ref2); } - - if (queue != null) - { - queue.deliverAsync(executor); - } + } public void stop() throws Exception @@ -474,9 +470,14 @@ public void connectionFailed(final HornetQException me) { - fail(); + fail(false); } + public void beforeReconnect(final HornetQException exception) + { + fail(true); + } + // Package protected --------------------------------------------- // Protected ----------------------------------------------------- @@ -498,48 +499,59 @@ } } - private void fail() + private void fail(final boolean beforeReconnect) { - //This will get called even after the bridge reconnects - in this case - //we want to cancel all unacked refs so they get resent - //duplicate detection will ensure no dups are routed on the other side - + // This will get called even after the bridge reconnects - in this case + // we want to cancel all unacked refs so they get resent + // duplicate detection will ensure no dups are routed on the other side + if (session.getConnection().isDestroyed()) { active = false; } - + try { if (!session.getConnection().isDestroyed()) { - setupNotificationConsumer(); - - cancelRefs(); + if (beforeReconnect) + { + active = false; + cancelRefs(); + } + else + { + setupNotificationConsumer(); + active = true; + if (queue != null) + { + queue.deliverAsync(executor); + } + } } } catch (Exception e) { log.error("Failed to cancel refs", e); } - } - + } + private ClientConsumer notifConsumer; - + // TODO - we should move this code to the ClusterConnectorImpl - and just execute it when the bridge // connection is opened and closed - we can use // a callback to tell us that private void setupNotificationConsumer() throws Exception - { + { if (flowRecord != null) { - + if (notifConsumer != null) { try { notifConsumer.close(); - + notifConsumer = null; } catch (HornetQException e) @@ -547,7 +559,7 @@ log.error("Failed to close consumer", e); } } - + // Get the queue data // Create a queue to catch the notifications - the name must be deterministic on live and backup, but @@ -559,30 +571,30 @@ SimpleString notifQueueName = new SimpleString(qName); SimpleString filter = new SimpleString(ManagementHelper.HDR_BINDING_TYPE + "<>" + - BindingType.DIVERT.toInt() + - " AND " + - ManagementHelper.HDR_NOTIFICATION_TYPE + - " IN ('" + - NotificationType.BINDING_ADDED + - "','" + - NotificationType.BINDING_REMOVED + - "','" + - NotificationType.CONSUMER_CREATED + - "','" + - NotificationType.CONSUMER_CLOSED + - "','" + - NotificationType.PROPOSAL + - "','" + - NotificationType.PROPOSAL_RESPONSE + - "') AND " + - ManagementHelper.HDR_DISTANCE + - "<" + - flowRecord.getMaxHops() + - " AND (" + - ManagementHelper.HDR_ADDRESS + - " LIKE '" + - flowRecord.getAddress() + - "%')"); + BindingType.DIVERT.toInt() + + " AND " + + ManagementHelper.HDR_NOTIFICATION_TYPE + + " IN ('" + + NotificationType.BINDING_ADDED + + "','" + + NotificationType.BINDING_REMOVED + + "','" + + NotificationType.CONSUMER_CREATED + + "','" + + NotificationType.CONSUMER_CLOSED + + "','" + + NotificationType.PROPOSAL + + "','" + + NotificationType.PROPOSAL_RESPONSE + + "') AND " + + ManagementHelper.HDR_DISTANCE + + "<" + + flowRecord.getMaxHops() + + " AND (" + + ManagementHelper.HDR_ADDRESS + + " LIKE '" + + flowRecord.getAddress() + + "%')"); // The queue can't be temporary, since if the node with the bridge crashes then is restarted quickly // it might get deleted on the target when it does connection cleanup @@ -632,7 +644,7 @@ } try - { + { if (discoveryAddress != null) { csf = new ClientSessionFactoryImpl(discoveryAddress, discoveryPort); @@ -664,19 +676,19 @@ session.setSendAcknowledgementHandler(BridgeImpl.this); setupNotificationConsumer(); - + active = true; - + queue.addConsumer(BridgeImpl.this); queue.deliverAsync(executor); - + return true; } catch (Exception e) { log.warn("Bridge " + name + " is unable to connect to destination. It will be disabled."); - + return false; } } @@ -710,6 +722,11 @@ queue.removeConsumer(BridgeImpl.this); cancelRefs(); + + if (queue != null) + { + queue.deliverAsync(executor); + } } catch (Exception e) { @@ -730,41 +747,41 @@ } } } - -// private class FailRunnable implements Runnable -// { -// public void run() -// { -// synchronized (BridgeImpl.this) -// { -// if (!started) -// { -// return; -// } -// -// active = false; -// } -// -// try -// { -// queue.removeConsumer(BridgeImpl.this); -// -// session.cleanUp(); -// -// cancelRefs(); -// -// csf.close(); -// } -// catch (Exception e) -// { -// log.error("Failed to stop", e); -// } -// -// if (!createObjects()) -// { -// started = false; -// } -// } -// } + // private class FailRunnable implements Runnable + // { + // public void run() + // { + // synchronized (BridgeImpl.this) + // { + // if (!started) + // { + // return; + // } + // + // active = false; + // } + // + // try + // { + // queue.removeConsumer(BridgeImpl.this); + // + // session.cleanUp(); + // + // cancelRefs(); + // + // csf.close(); + // } + // catch (Exception e) + // { + // log.error("Failed to stop", e); + // } + // + // if (!createObjects()) + // { + // started = false; + // } + // } + // } + } Modified: trunk/src/main/org/hornetq/core/server/cluster/impl/Redistributor.java =================================================================== --- trunk/src/main/org/hornetq/core/server/cluster/impl/Redistributor.java 2009-10-30 15:12:21 UTC (rev 8170) +++ trunk/src/main/org/hornetq/core/server/cluster/impl/Redistributor.java 2009-10-30 17:41:19 UTC (rev 8171) @@ -123,7 +123,7 @@ final Transaction tx = new TransactionImpl(storageManager); - boolean routed = postOffice.redistribute(reference.getMessage(), queue, new RoutingContextImpl(tx)); + boolean routed = postOffice.redistribute(reference.getMessage(), queue, tx); if (routed) { Modified: trunk/src/main/org/hornetq/core/server/impl/DivertImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/server/impl/DivertImpl.java 2009-10-30 15:12:21 UTC (rev 8170) +++ trunk/src/main/org/hornetq/core/server/impl/DivertImpl.java 2009-10-30 17:41:19 UTC (rev 8171) @@ -77,17 +77,24 @@ public void route(ServerMessage message, final RoutingContext context) throws Exception { SimpleString originalDestination = message.getDestination(); - - message.setDestination(forwardAddress); + + //We must make a copy of the message, otherwise things like returning credits to the page won't work + //properly on ack, since the original destination will be overwritten - message.putStringProperty(HDR_ORIGINAL_DESTINATION, originalDestination); + //TODO we can optimise this so it doesn't copy if it's not routed anywhere else + + ServerMessage copy = message.copy(); + + copy.setDestination(forwardAddress); + + copy.putStringProperty(HDR_ORIGINAL_DESTINATION, originalDestination); if (transformer != null) { - message = transformer.transform(message); + copy = transformer.transform(copy); } - postOffice.route(message, context); + postOffice.route(copy, context.getTransaction()); } public SimpleString getRoutingName() Modified: trunk/src/main/org/hornetq/core/server/impl/HornetQPacketHandler.java =================================================================== --- trunk/src/main/org/hornetq/core/server/impl/HornetQPacketHandler.java 2009-10-30 15:12:21 UTC (rev 8170) +++ trunk/src/main/org/hornetq/core/server/impl/HornetQPacketHandler.java 2009-10-30 17:41:19 UTC (rev 8171) @@ -189,17 +189,19 @@ } } - private void handleCreateReplication(CreateReplicationSessionMessage request) + private void handleCreateReplication(final CreateReplicationSessionMessage request) { Packet response; try { - Channel channel = connection.getChannel(request.getSessionChannelID(), request.getWindowSize(), false); + Channel channel = connection.getChannel(request.getSessionChannelID(), request.getWindowSize()); + ReplicationEndpoint endpoint = server.createReplicationEndpoint(channel); + channel.setHandler(endpoint); + response = new NullResponseMessage(); - } catch (Exception e) { Modified: trunk/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java 2009-10-30 15:12:21 UTC (rev 8170) +++ trunk/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java 2009-10-30 17:41:19 UTC (rev 8171) @@ -329,7 +329,7 @@ clusterManager.stop(); } - if(groupingHandler != null) + if (groupingHandler != null) { managementService.removeNotificationListener(groupingHandler); groupingHandler = null; @@ -536,10 +536,32 @@ } else { - // Reconnect the channel to the new connection - int serverLastReceivedCommandID = session.transferConnection(connection, lastReceivedCommandID); + if (session.getChannel().getConfirmationWindowSize() == -1) + { + // Even though session exists, we can't reattach since confi window size == -1, + // i.e. we don't have a resend cache for commands, so we just close the old session + // and let the client recreate - return new ReattachSessionResponseMessage(serverLastReceivedCommandID, true); + try + { + session.close(); + } + catch (Exception e) + { + log.error("Failed to close session", e); + } + + sessions.remove(name); + + return new ReattachSessionResponseMessage(-1, false); + } + else + { + // Reconnect the channel to the new connection + int serverLastReceivedCommandID = session.transferConnection(connection, lastReceivedCommandID); + + return new ReattachSessionResponseMessage(serverLastReceivedCommandID, true); + } } } @@ -590,7 +612,7 @@ currentSession.getChannel().close(); } - Channel channel = connection.getChannel(channelID, sendWindowSize, false); + Channel channel = connection.getChannel(channelID, sendWindowSize); final ServerSessionImpl session = new ServerSessionImpl(name, username, @@ -878,7 +900,6 @@ return true; } - private synchronized void callActivateCallbacks() { for (ActivateCallback callback : activateCallbacks) @@ -1162,9 +1183,11 @@ for (GroupingInfo groupingInfo : groupingInfos) { - if(groupingHandler != null) + if (groupingHandler != null) { - groupingHandler.addGroupBinding(new GroupBinding(groupingInfo.getId(), groupingInfo.getGroupId(), groupingInfo.getClusterName())); + groupingHandler.addGroupBinding(new GroupBinding(groupingInfo.getId(), + groupingInfo.getGroupId(), + groupingInfo.getClusterName())); } } @@ -1331,13 +1354,18 @@ GroupingHandler groupingHandler; if (config.getType() == GroupingHandlerConfiguration.TYPE.LOCAL) { - groupingHandler = new LocalGroupingHandler(managementService, config.getName(), config.getAddress(), - getStorageManager(), - config.getTimeout()); + groupingHandler = new LocalGroupingHandler(managementService, + config.getName(), + config.getAddress(), + getStorageManager(), + config.getTimeout()); } else { - groupingHandler = new RemoteGroupingHandler(managementService, config.getName(), config.getAddress(), config.getTimeout()); + groupingHandler = new RemoteGroupingHandler(managementService, + config.getName(), + config.getAddress(), + config.getTimeout()); } log.info("deploying grouping handler: " + groupingHandler); this.groupingHandler = groupingHandler; Modified: trunk/src/main/org/hornetq/core/server/impl/QueueImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/server/impl/QueueImpl.java 2009-10-30 15:12:21 UTC (rev 8170) +++ trunk/src/main/org/hornetq/core/server/impl/QueueImpl.java 2009-10-30 17:41:19 UTC (rev 8171) @@ -51,6 +51,7 @@ import org.hornetq.core.server.ServerMessage; import org.hornetq.core.server.cluster.impl.Redistributor; import org.hornetq.core.settings.HierarchicalRepository; +import org.hornetq.core.settings.impl.AddressFullMessagePolicy; import org.hornetq.core.settings.impl.AddressSettings; import org.hornetq.core.transaction.Transaction; import org.hornetq.core.transaction.TransactionOperation; @@ -210,7 +211,7 @@ { return false; } - + public void route(final ServerMessage message, final RoutingContext context) throws Exception { context.addQueue(this); @@ -261,7 +262,7 @@ } public void addLast(final MessageReference ref) - { + { add(ref, false); } @@ -947,7 +948,7 @@ copyMessage.setDestination(toAddress); - postOffice.route(copyMessage, new RoutingContextImpl(tx)); + postOffice.route(copyMessage, tx); acknowledge(tx, ref); } @@ -967,7 +968,7 @@ long newID = storageManager.generateUniqueID(); ServerMessage copy = message.makeCopyForExpiryOrDLA(newID, expiry); - + return copy; } @@ -1034,7 +1035,7 @@ copyMessage.setDestination(address); - postOffice.route(copyMessage, new RoutingContextImpl(tx)); + postOffice.route(copyMessage, tx); acknowledge(tx, ref); @@ -1110,7 +1111,7 @@ promptDelivery = false; return; } - + continue; } else @@ -1158,7 +1159,7 @@ } HandleStatus status = handle(reference, consumer); - + if (status == HandleStatus.HANDLED) { if (iterator == null) @@ -1409,25 +1410,20 @@ queue.deliveringCount.decrementAndGet(); - // TODO: We could optimize this by storing the paging-store for the address on the Queue. We would need to know - // the Address for the Queue PagingStore store; - if (pagingManager != null) { + // TODO: We could optimize this by storing the paging-store for the address on the Queue. We would need to know + // the Address for the Queue store = pagingManager.getPageStore(ref.getMessage().getDestination()); - - store.addSize(-ref.getMemoryEstimate()); } else { store = null; } - if (message.decrementRefCount() == 0 && store != null) - { - store.addSize(-ref.getMessage().getMemoryEstimate()); - } + message.decrementRefCount(store, ref); + } void postRollback(LinkedList<MessageReference> refs) throws Exception @@ -1435,10 +1431,10 @@ synchronized (this) { direct = false; - + for (MessageReference ref : refs) { - add(ref, true); + add(ref, true); } deliver(); @@ -1467,7 +1463,7 @@ if (pagingStore != null) { // If the queue is empty, we need to check if there are pending messages, and throw a warning - if (pagingStore.isPaging() && !pagingStore.isDropWhenMaxSize()) + if (pagingStore.isPaging() && pagingStore.getAddressFullMessagePolicy() == AddressFullMessagePolicy.PAGE) { // This is just a *request* to depage. Depage will only happens if there is space on the Address // and GlobalSize @@ -1603,7 +1599,7 @@ public synchronized void resume() { paused = false; - + deliver(); } Modified: trunk/src/main/org/hornetq/core/server/impl/ServerConsumerImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/server/impl/ServerConsumerImpl.java 2009-10-30 15:12:21 UTC (rev 8170) +++ trunk/src/main/org/hornetq/core/server/impl/ServerConsumerImpl.java 2009-10-30 17:41:19 UTC (rev 8171) @@ -29,8 +29,6 @@ import org.hornetq.core.management.ManagementService; import org.hornetq.core.management.Notification; import org.hornetq.core.management.NotificationType; -import org.hornetq.core.paging.PagingManager; -import org.hornetq.core.paging.PagingStore; import org.hornetq.core.persistence.StorageManager; import org.hornetq.core.postoffice.Binding; import org.hornetq.core.postoffice.QueueBinding; @@ -110,8 +108,6 @@ private final StorageManager storageManager; - private final PagingManager pagingManager; - private final java.util.Queue<MessageReference> deliveringRefs = new ConcurrentLinkedQueue<MessageReference>(); private final Channel channel; @@ -134,7 +130,6 @@ final boolean started, final boolean browseOnly, final StorageManager storageManager, - final PagingManager pagingManager, final Channel channel, final boolean preAcknowledge, final boolean updateDeliveries, @@ -164,8 +159,6 @@ this.preAcknowledge = preAcknowledge; - this.pagingManager = pagingManager; - this.managementService = managementService; this.minLargeMessageSize = session.getMinLargeMessageSize(); @@ -206,7 +199,7 @@ if (largeMessageDeliverer != null) { - largeMessageDeliverer.close(); + largeMessageDeliverer.finish(); } if (!browseOnly) @@ -546,7 +539,7 @@ if (message.isLargeMessage()) { // we must hold one reference, or the file will be deleted before it could be delivered - message.incrementRefCount(); + ((LargeServerMessage)message).incrementDelayDeletionCount(); } // With pre-ack, we ack *before* sending to the client @@ -573,6 +566,7 @@ } private void deliverLargeMessage(final MessageReference ref, final ServerMessage message) + throws Exception { pendingLargeMessagesCounter.incrementAndGet(); @@ -622,6 +616,10 @@ } } } + catch (Exception e) + { + log.error("Failed to run large message deliverer", e); + } finally { lock.unlock(); @@ -646,18 +644,18 @@ private volatile long positionPendingLargeMessage; public LargeMessageDeliverer(final LargeServerMessage message, final MessageReference ref) + throws Exception { pendingLargeMessage = message; - // we must hold one reference, or the file will be deleted before it could be delivered - pendingLargeMessage.incrementRefCount(); + pendingLargeMessage.incrementDelayDeletionCount(); sizePendingLargeMessage = pendingLargeMessage.getLargeBodySize(); this.ref = ref; } - public boolean deliver() + public boolean deliver() throws Exception { lock.lock(); @@ -672,6 +670,7 @@ { return false; } + SessionReceiveMessage initialMessage; if (sentFirstMessage) @@ -682,9 +681,9 @@ { sentFirstMessage = true; - HornetQBuffer headerBuffer = ChannelBuffers.buffer(pendingLargeMessage.getPropertiesEncodeSize()); + HornetQBuffer headerBuffer = ChannelBuffers.buffer(pendingLargeMessage.getHeadersAndPropertiesEncodeSize()); - pendingLargeMessage.encodeProperties(headerBuffer); + pendingLargeMessage.encodeHeadersAndProperties(headerBuffer); initialMessage = new SessionReceiveMessage(id, headerBuffer.array(), @@ -697,6 +696,8 @@ if (availableCredits != null) { // Flow control needs to be done in advance. + + //Again WHY? Is this necessary now we don't replicate sessions? precalculateAvailableCredits = preCalculateFlowControl(initialMessage); } else @@ -759,7 +760,7 @@ trace("Finished deliverLargeMessage"); } - close(); + finish(); return true; } @@ -772,39 +773,18 @@ /** * */ - public void close() + public void finish() throws Exception { pendingLargeMessage.releaseResources(); + + pendingLargeMessage.decrementDelayDeletionCount(); - int counter = pendingLargeMessage.decrementRefCount(); - if (preAcknowledge && !browseOnly) { // PreAck will have an extra reference - counter = pendingLargeMessage.decrementRefCount(); + pendingLargeMessage.decrementDelayDeletionCount(); } - if (!browseOnly) - { - // We added a reference to avoid deleting the file before it was delivered - // if (pendingLargeMessage.decrementRefCount() == 0) - if (counter == 0) - { - // The decrement was deferred to large-message, hence we need to - // subtract the size inside largeMessage - try - { - PagingStore store = pagingManager.getPageStore(binding.getAddress()); - store.addSize(-pendingLargeMessage.getMemoryEstimate()); - } - catch (Exception e) - { - // This shouldn't happen as the pageStore should have been initialized already. - log.error("Error getting pageStore", e); - } - } - } - largeMessageDeliverer = null; pendingLargeMessagesCounter.decrementAndGet(); Modified: trunk/src/main/org/hornetq/core/server/impl/ServerMessageImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/server/impl/ServerMessageImpl.java 2009-10-30 15:12:21 UTC (rev 8170) +++ trunk/src/main/org/hornetq/core/server/impl/ServerMessageImpl.java 2009-10-30 17:41:19 UTC (rev 8171) @@ -17,6 +17,7 @@ import org.hornetq.core.logging.Logger; import org.hornetq.core.message.impl.MessageImpl; +import org.hornetq.core.paging.PagingStore; import org.hornetq.core.remoting.spi.HornetQBuffer; import org.hornetq.core.server.MessageReference; import org.hornetq.core.server.Queue; @@ -111,10 +112,39 @@ stored = true; } - public int incrementRefCount() + public int incrementRefCount(final PagingStore pagingStore, final MessageReference reference) throws Exception { - return refCount.incrementAndGet(); + int count = refCount.incrementAndGet(); + + if (pagingStore != null) + { + if (count == 1) + { + pagingStore.addSize(this, true); + } + + pagingStore.addSize(reference, true); + } + + return count; } + + public int decrementRefCount(final PagingStore pagingStore, final MessageReference reference) throws Exception + { + int count = refCount.decrementAndGet(); + + if (pagingStore != null) + { + if (count == 0) + { + pagingStore.addSize(this, false); + } + + pagingStore.addSize(reference, false); + } + + return count; + } public int incrementDurableRefCount() { @@ -126,10 +156,7 @@ return durableRefCount.decrementAndGet(); } - public int decrementRefCount() - { - return refCount.decrementAndGet(); - } + public int getRefCount() { @@ -170,9 +197,7 @@ public ServerMessage copy() throws Exception { - ServerMessage m = new ServerMessageImpl(this); - - return m; + return new ServerMessageImpl(this); } public ServerMessage makeCopyForExpiryOrDLA(final long newID, final boolean expiry) throws Exception Added: trunk/src/main/org/hornetq/core/server/impl/ServerProducerCreditManager.java =================================================================== --- trunk/src/main/org/hornetq/core/server/impl/ServerProducerCreditManager.java (rev 0) +++ trunk/src/main/org/hornetq/core/server/impl/ServerProducerCreditManager.java 2009-10-30 17:41:19 UTC (rev 8171) @@ -0,0 +1,35 @@ +/* + * 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 tim + * + * + */ +public interface ServerProducerCreditManager +{ + int creditsReleased(int credits); + + int acquireCredits(int credits, CreditsAvailableRunnable runnable); + + int waitingEntries(); + + PagingStore getStore(); +} Added: trunk/src/main/org/hornetq/core/server/impl/ServerProducerCreditManagerImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/server/impl/ServerProducerCreditManagerImpl.java (rev 0) +++ trunk/src/main/org/hornetq/core/server/impl/ServerProducerCreditManagerImpl.java 2009-10-30 17:41:19 UTC (rev 8171) @@ -0,0 +1,145 @@ +/* + * 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 2009-10-30 15:12:21 UTC (rev 8170) +++ trunk/src/main/org/hornetq/core/server/impl/ServerSessionImpl.java 2009-10-30 17:41:19 UTC (rev 8171) @@ -16,6 +16,7 @@ import static org.hornetq.core.management.NotificationType.CONSUMER_CREATED; import java.util.ArrayList; +import java.util.HashMap; import java.util.HashSet; import java.util.List; import java.util.Map; @@ -27,7 +28,6 @@ import javax.transaction.xa.XAResource; import javax.transaction.xa.Xid; -import org.hornetq.core.buffers.ChannelBuffers; import org.hornetq.core.client.impl.ClientMessageImpl; import org.hornetq.core.client.management.impl.ManagementHelper; import org.hornetq.core.exception.HornetQException; @@ -36,6 +36,7 @@ import org.hornetq.core.logging.Logger; import org.hornetq.core.management.ManagementService; import org.hornetq.core.management.Notification; +import org.hornetq.core.paging.PagingStore; import org.hornetq.core.persistence.StorageManager; import org.hornetq.core.postoffice.Binding; import org.hornetq.core.postoffice.BindingType; @@ -56,13 +57,15 @@ import org.hornetq.core.remoting.impl.wireformat.SessionBindingQueryMessage; import org.hornetq.core.remoting.impl.wireformat.SessionBindingQueryResponseMessage; import org.hornetq.core.remoting.impl.wireformat.SessionConsumerCloseMessage; -import org.hornetq.core.remoting.impl.wireformat.SessionForceConsumerDelivery; import org.hornetq.core.remoting.impl.wireformat.SessionConsumerFlowCreditMessage; import org.hornetq.core.remoting.impl.wireformat.SessionCreateConsumerMessage; import org.hornetq.core.remoting.impl.wireformat.SessionDeleteQueueMessage; import org.hornetq.core.remoting.impl.wireformat.SessionExpiredMessage; +import org.hornetq.core.remoting.impl.wireformat.SessionForceConsumerDelivery; +import org.hornetq.core.remoting.impl.wireformat.SessionProducerCreditsMessage; import org.hornetq.core.remoting.impl.wireformat.SessionQueueQueryMessage; import org.hornetq.core.remoting.impl.wireformat.SessionQueueQueryResponseMessage; +import org.hornetq.core.remoting.impl.wireformat.SessionRequestProducerCreditsMessage; import org.hornetq.core.remoting.impl.wireformat.SessionSendContinuationMessage; import org.hornetq.core.remoting.impl.wireformat.SessionSendLargeMessage; import org.hornetq.core.remoting.impl.wireformat.SessionSendMessage; @@ -79,7 +82,6 @@ import org.hornetq.core.remoting.impl.wireformat.SessionXASetTimeoutMessage; import org.hornetq.core.remoting.impl.wireformat.SessionXASetTimeoutResponseMessage; import org.hornetq.core.remoting.impl.wireformat.SessionXAStartMessage; -import org.hornetq.core.remoting.spi.HornetQBuffer; import org.hornetq.core.security.CheckType; import org.hornetq.core.security.SecurityStore; import org.hornetq.core.server.HornetQServer; @@ -183,7 +185,7 @@ final SecurityStore securityStore, final Executor executor, final Channel channel, - final ManagementService managementService, + final ManagementService managementService, final HornetQServer server, final SimpleString managementAddress) throws Exception { @@ -285,7 +287,9 @@ } } - public void close() throws Exception + private boolean closed; + + public synchronized void close() throws Exception { if (tx != null && tx.getXid() == null) { @@ -318,6 +322,15 @@ } remotingConnection.removeFailureListener(this); + + // Return any outstanding credits + + closed = true; + + for (CreditManagerHolder holder : creditManagerHolders.values()) + { + holder.store.returnProducerCredits(holder.outstandingCredits); + } } public void promptDelivery(final Queue queue, boolean async) @@ -362,7 +375,6 @@ started, browseOnly, storageManager, - postOffice.getPagingManager(), channel, preAcknowledge, updateDeliveries, @@ -623,7 +635,7 @@ sendResponse(packet, response, false, false); } - + public void handleForceConsumerDelivery(SessionForceConsumerDelivery message) { try @@ -691,7 +703,6 @@ log.error("Failed to acknowledge", e); } - sendResponse(packet, null, false, false); } @@ -776,16 +787,22 @@ // checked heuristic committed transactions if (resourceManager.getHeuristicCommittedTransactions().contains(xid)) { - response = new SessionXAResponseMessage(true, XAException.XA_HEURCOM, "transaction has been heuristically committed: " + xid); + response = new SessionXAResponseMessage(true, + XAException.XA_HEURCOM, + "transaction has been heuristically committed: " + xid); } // checked heuristic rolled back transactions else if (resourceManager.getHeuristicRolledbackTransactions().contains(xid)) { - response = new SessionXAResponseMessage(true, XAException.XA_HEURRB, "transaction has been heuristically rolled back: " + xid); - } + response = new SessionXAResponseMessage(true, + XAException.XA_HEURRB, + "transaction has been heuristically rolled back: " + xid); + } else { - response = new SessionXAResponseMessage(true, XAException.XAER_NOTA, "Cannot find xid in resource manager: " + xid); + response = new SessionXAResponseMessage(true, + XAException.XAER_NOTA, + "Cannot find xid in resource manager: " + xid); } } else @@ -912,11 +929,12 @@ e.printStackTrace(); code = XAException.XAER_RMERR; } - } else + } + else { code = XAException.XAER_NOTA; } - + Packet response = new SessionXAResponseMessage((code != XAResource.XA_OK), code, null); sendResponse(packet, response, false, false); @@ -1054,16 +1072,22 @@ // checked heuristic committed transactions if (resourceManager.getHeuristicCommittedTransactions().contains(xid)) { - response = new SessionXAResponseMessage(true, XAException.XA_HEURCOM, "transaction has ben heuristically committed: " + xid); + response = new SessionXAResponseMessage(true, + XAException.XA_HEURCOM, + "transaction has ben heuristically committed: " + xid); } // checked heuristic rolled back transactions else if (resourceManager.getHeuristicRolledbackTransactions().contains(xid)) { - response = new SessionXAResponseMessage(true, XAException.XA_HEURRB, "transaction has ben heuristically rolled back: " + xid); - } + response = new SessionXAResponseMessage(true, + XAException.XA_HEURRB, + "transaction has ben heuristically rolled back: " + xid); + } else { - response = new SessionXAResponseMessage(true, XAException.XAER_NOTA, "Cannot find xid in resource manager: " + xid); + response = new SessionXAResponseMessage(true, + XAException.XAER_NOTA, + "Cannot find xid in resource manager: " + xid); } } else @@ -1264,7 +1288,7 @@ indoubtsXids.addAll(resourceManager.getHeuristicCommittedTransactions()); indoubtsXids.addAll(resourceManager.getHeuristicRolledbackTransactions()); Packet response = new SessionXAGetInDoubtXidsResponseMessage(indoubtsXids); - + sendResponse(packet, response, false, false); } @@ -1381,7 +1405,6 @@ log.error("Failed to receive credits " + this.server.getConfiguration().isBackup(), e); } - sendResponse(packet, null, false, false); } @@ -1392,33 +1415,30 @@ final LargeServerMessage msg = doCreateLargeMessage(id, packet); - if (msg == null) + if (msg != null) { - // packet logged an error, and played with channel.returns... and nothing needs to be done now - return; - } + // With a send we must make sure it is replicated to backup before being processed on live + // or can end up with delivery being processed on backup before original send - // With a send we must make sure it is replicated to backup before being processed on live - // or can end up with delivery being processed on backup before original send + if (currentLargeMessage != null) + { + log.warn("Replacing incomplete LargeMessage with ID=" + currentLargeMessage.getMessageID()); + } - if (currentLargeMessage != null) - { - log.warn("Replacing incomplete LargeMessage with ID=" + currentLargeMessage.getMessageID()); + currentLargeMessage = msg; + + sendResponse(packet, null, false, false); } - - currentLargeMessage = msg; - - sendResponse(packet, null, false, false); } public void handleSend(final SessionSendMessage packet) { Packet response = null; + ServerMessage message = packet.getServerMessage(); + try - { - ServerMessage message = packet.getServerMessage(); - + { long id = storageManager.generateUniqueID(); message.setMessageID(id); @@ -1455,7 +1475,18 @@ } } } - + finally + { + try + { + releaseOutStanding(message); + } + catch (Exception e) + { + log.error("Failed to release outstanding credits", e); + } + } + sendResponse(packet, response, false, false); } @@ -1479,8 +1510,10 @@ currentLargeMessage = null; message.releaseResources(); - + send(message); + + releaseOutStanding(message); } if (packet.isRequiresResponse()) @@ -1508,10 +1541,89 @@ sendResponse(packet, response, false, false); } + private static final class CreditManagerHolder + { + CreditManagerHolder(final PagingStore store) + { + this.store = store; + + this.manager = store.getProducerCreditManager(); + } + + final PagingStore store; + + final ServerProducerCreditManager manager; + + volatile int outstandingCredits; + } + + private Map<SimpleString, CreditManagerHolder> creditManagerHolders = new HashMap<SimpleString, CreditManagerHolder>(); + + 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; + } + + public void handleRequestProducerCredits(final SessionRequestProducerCreditsMessage packet) throws Exception + { + final SimpleString address = packet.getAddress(); + + final CreditManagerHolder holder = this.getCreditManagerHolder(address); + + int credits = packet.getCredits(); + + int gotCredits = holder.manager.acquireCredits(credits, new CreditsAvailableRunnable() + { + public boolean run(int credits) + { + synchronized (ServerSessionImpl.this) + { + if (!closed) + { + sendProducerCredits(holder, credits, address); + + return true; + } + else + { + return false; + } + } + } + }); + + if (gotCredits > 0) + { + sendProducerCredits(holder, gotCredits, address); + } + + sendResponse(packet, null, false, false); + } + + private void sendProducerCredits(final CreditManagerHolder holder, final int credits, final SimpleString address) + { + holder.outstandingCredits += credits; + + Packet packet = new SessionProducerCreditsMessage(credits, address); + + channel.send(packet); + } + public int transferConnection(final RemotingConnection newConnection, final int lastReceivedCommandID) { boolean wasStarted = this.started; - + if (wasStarted) { this.setStarted(false); @@ -1528,7 +1640,7 @@ // received responses that the backup did not know about. channel.transferConnection(newConnection); - + newConnection.syncIDGeneratorSequence(remotingConnection.getIDGeneratorSequence()); remotingConnection = newConnection; @@ -1544,7 +1656,7 @@ { this.setStarted(true); } - + return serverLastReceivedCommandID; } @@ -1618,7 +1730,10 @@ * @param packet * @param response */ - private void sendResponse(final Packet confirmPacket, final Packet response, final boolean flush, final boolean closeChannel) + private void sendResponse(final Packet confirmPacket, + final Packet response, + final boolean flush, + final boolean closeChannel) { if (storageManager.isReplicated()) { @@ -1637,7 +1752,7 @@ doSendResponse(confirmPacket, response, flush, closeChannel); } } - + /** * @param confirmPacket * @param response @@ -1662,14 +1777,13 @@ { channel.send(response); } - + if (closeChannel) { channel.close(); } } - private void setStarted(final boolean s) { Set<ServerConsumer> consumersClone = new HashSet<ServerConsumer>(consumers.values()); @@ -1785,6 +1899,28 @@ tx = new TransactionImpl(storageManager); } + + /* + * 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) throws Exception + { + CreditManagerHolder holder = getCreditManagerHolder(message.getDestination()); + + int size = message.getEncodeSize(); + + holder.outstandingCredits -= size; + + holder.store.returnProducerCredits(size); + } private void send(final ServerMessage msg) throws Exception { @@ -1801,14 +1937,14 @@ } throw e; } - + if (tx == null || autoCommitSends) { - postOffice.route(msg); + postOffice.route(msg); } else { - postOffice.route(msg, new RoutingContextImpl(tx)); - } + postOffice.route(msg, tx); + } } } Modified: trunk/src/main/org/hornetq/core/server/impl/ServerSessionPacketHandler.java =================================================================== --- trunk/src/main/org/hornetq/core/server/impl/ServerSessionPacketHandler.java 2009-10-30 15:12:21 UTC (rev 8170) +++ trunk/src/main/org/hornetq/core/server/impl/ServerSessionPacketHandler.java 2009-10-30 17:41:19 UTC (rev 8171) @@ -48,6 +48,7 @@ import org.hornetq.core.remoting.ChannelHandler; import org.hornetq.core.remoting.Packet; import org.hornetq.core.remoting.impl.wireformat.CreateQueueMessage; +import org.hornetq.core.remoting.impl.wireformat.PacketImpl; import org.hornetq.core.remoting.impl.wireformat.RollbackMessage; import org.hornetq.core.remoting.impl.wireformat.SessionAcknowledgeMessage; import org.hornetq.core.remoting.impl.wireformat.SessionBindingQueryMessage; @@ -58,6 +59,7 @@ import org.hornetq.core.remoting.impl.wireformat.SessionDeleteQueueMessage; import org.hornetq.core.remoting.impl.wireformat.SessionExpiredMessage; import org.hornetq.core.remoting.impl.wireformat.SessionQueueQueryMessage; +import org.hornetq.core.remoting.impl.wireformat.SessionRequestProducerCreditsMessage; import org.hornetq.core.remoting.impl.wireformat.SessionSendContinuationMessage; import org.hornetq.core.remoting.impl.wireformat.SessionSendLargeMessage; import org.hornetq.core.remoting.impl.wireformat.SessionSendMessage; @@ -275,6 +277,13 @@ session.handleForceConsumerDelivery(message); break; } + case PacketImpl.SESS_PRODUCER_REQUEST_CREDITS: + { + SessionRequestProducerCreditsMessage message = (SessionRequestProducerCreditsMessage) + packet; + session.handleRequestProducerCredits(message); + break; + } } } catch (Throwable t) Modified: trunk/src/main/org/hornetq/core/settings/impl/AddressSettings.java =================================================================== --- trunk/src/main/org/hornetq/core/settings/impl/AddressSettings.java 2009-10-30 15:12:21 UTC (rev 8170) +++ trunk/src/main/org/hornetq/core/settings/impl/AddressSettings.java 2009-10-30 17:41:19 UTC (rev 8171) @@ -36,10 +36,10 @@ public static final int DEFAULT_MAX_SIZE_BYTES = -1; - public static final boolean DEFAULT_DROP_MESSAGES_WHEN_FULL = false; + public static final AddressFullMessagePolicy DEFAULT_ADDRESS_FULL_MESSAGE_POLICY = AddressFullMessagePolicy.PAGE; public static final int DEFAULT_PAGE_SIZE = 10 * 1024 * 1024; - + public static final int DEFAULT_MAX_DELIVERY_ATTEMPTS = 10; public static final int DEFAULT_MESSAGE_COUNTER_HISTORY_DAY_LIMIT = 0; @@ -49,9 +49,11 @@ public static final boolean DEFAULT_LAST_VALUE_QUEUE = false; public static final long DEFAULT_REDISTRIBUTION_DELAY = -1; - + public static final boolean DEFAULT_SEND_TO_DLA_ON_NO_ROUTE = false; + private AddressFullMessagePolicy addressFullMessagePolicy = null; + private Long maxSizeBytes = null; private Integer pageSizeBytes = null; @@ -73,7 +75,7 @@ private Boolean lastValueQueue = null; private Long redistributionDelay = null; - + private Boolean sendToDLAOnNoRoute = null; public boolean isLastValueQueue() @@ -86,21 +88,21 @@ this.lastValueQueue = lastValueQueue; } - public int getPageSizeBytes() + public AddressFullMessagePolicy getAddressFullMessagePolicy() { - return pageSizeBytes != null ? pageSizeBytes : DEFAULT_PAGE_SIZE; + return addressFullMessagePolicy != null ? addressFullMessagePolicy : DEFAULT_ADDRESS_FULL_MESSAGE_POLICY; } - public boolean isDropMessagesWhenFull() + public void setAddressFullMessagePolicy(final AddressFullMessagePolicy addressFullMessagePolicy) { - return dropMessagesWhenFull != null ? dropMessagesWhenFull : DEFAULT_DROP_MESSAGES_WHEN_FULL; + this.addressFullMessagePolicy = addressFullMessagePolicy; } - - public void setDropMessagesWhenFull(final boolean value) + + public int getPageSizeBytes() { - dropMessagesWhenFull = value; + return pageSizeBytes != null ? pageSizeBytes : DEFAULT_PAGE_SIZE; } - + public void setPageSizeBytes(final int pageSize) { pageSizeBytes = pageSize; @@ -176,7 +178,7 @@ { this.expiryAddress = expiryAddress; } - + public boolean isSendToDLAOnNoRoute() { return sendToDLAOnNoRoute != null ? sendToDLAOnNoRoute : DEFAULT_SEND_TO_DLA_ON_NO_ROUTE; @@ -266,6 +268,10 @@ { sendToDLAOnNoRoute = merged.sendToDLAOnNoRoute; } + if (addressFullMessagePolicy == null) + { + addressFullMessagePolicy = merged.addressFullMessagePolicy; + } } } Modified: trunk/src/main/org/hornetq/jms/client/HornetQConnection.java =================================================================== --- trunk/src/main/org/hornetq/jms/client/HornetQConnection.java 2009-10-30 15:12:21 UTC (rev 8170) +++ trunk/src/main/org/hornetq/jms/client/HornetQConnection.java 2009-10-30 17:41:19 UTC (rev 8171) @@ -41,9 +41,9 @@ import org.hornetq.core.client.ClientSession; import org.hornetq.core.client.ClientSessionFactory; +import org.hornetq.core.client.SessionFailureListener; import org.hornetq.core.exception.HornetQException; import org.hornetq.core.logging.Logger; -import org.hornetq.core.remoting.FailureListener; import org.hornetq.core.version.Version; import org.hornetq.utils.SimpleString; import org.hornetq.utils.UUIDGenerator; @@ -104,7 +104,7 @@ private final String password; - private final FailureListener listener = new JMSFailureListener(this); + private final SessionFailureListener listener = new JMSFailureListener(this); private final Version thisVersion; @@ -551,7 +551,7 @@ // Inner classes -------------------------------------------------------------------------------- - private static class JMSFailureListener implements FailureListener + private static class JMSFailureListener implements SessionFailureListener { private WeakReference<HornetQConnection> connectionRef; @@ -599,6 +599,10 @@ } } } + + public void beforeReconnect(final HornetQException me) + { + } } } Modified: trunk/src/main/org/hornetq/jms/client/HornetQConnectionFactory.java =================================================================== --- trunk/src/main/org/hornetq/jms/client/HornetQConnectionFactory.java 2009-10-30 15:12:21 UTC (rev 8170) +++ trunk/src/main/org/hornetq/jms/client/HornetQConnectionFactory.java 2009-10-30 17:41:19 UTC (rev 8171) @@ -330,14 +330,14 @@ sessionFactory.setConsumerMaxRate(consumerMaxRate); } - public synchronized int getProducerWindowSize() + public synchronized int getConfirmationWindowSize() { - return sessionFactory.getProducerWindowSize(); + return sessionFactory.getConfirmationWindowSize(); } - public synchronized void setProducerWindowSize(int producerWindowSize) + public synchronized void setConfirmationWindowSize(int confirmationWindowSize) { - sessionFactory.setProducerWindowSize(producerWindowSize); + sessionFactory.setConfirmationWindowSize(confirmationWindowSize); } public synchronized int getProducerMaxRate() Modified: trunk/src/main/org/hornetq/jms/server/JMSServerManager.java =================================================================== --- trunk/src/main/org/hornetq/jms/server/JMSServerManager.java 2009-10-30 15:12:21 UTC (rev 8170) +++ trunk/src/main/org/hornetq/jms/server/JMSServerManager.java 2009-10-30 17:41:19 UTC (rev 8171) @@ -146,7 +146,7 @@ int minLargeMessageSize, int consumerWindowSize, int consumerMaxRate, - int producerWindowSize, + int confirmationWindowSize, int producerMaxRate, boolean blockOnAcknowledge, boolean blockOnPersistentSend, @@ -178,7 +178,7 @@ int minLargeMessageSize, int consumerWindowSize, int consumerMaxRate, - int producerWindowSize, + int confirmationWindowSize, int producerMaxRate, boolean blockOnAcknowledge, boolean blockOnPersistentSend, Modified: trunk/src/main/org/hornetq/jms/server/config/ConnectionFactoryConfiguration.java =================================================================== --- trunk/src/main/org/hornetq/jms/server/config/ConnectionFactoryConfiguration.java 2009-10-30 15:12:21 UTC (rev 8170) +++ trunk/src/main/org/hornetq/jms/server/config/ConnectionFactoryConfiguration.java 2009-10-30 17:41:19 UTC (rev 8171) @@ -79,9 +79,9 @@ void setConsumerMaxRate(int consumerMaxRate); - int getProducerWindowSize(); + int getConfirmationWindowSize(); - void setProducerWindowSize(int producerWindowSize); + void setConfirmationWindowSize(int confirmationWindowSize); int getProducerMaxRate(); Modified: trunk/src/main/org/hornetq/jms/server/config/impl/ConnectionFactoryConfigurationImpl.java =================================================================== --- trunk/src/main/org/hornetq/jms/server/config/impl/ConnectionFactoryConfigurationImpl.java 2009-10-30 15:12:21 UTC (rev 8170) +++ trunk/src/main/org/hornetq/jms/server/config/impl/ConnectionFactoryConfigurationImpl.java 2009-10-30 17:41:19 UTC (rev 8171) @@ -62,7 +62,7 @@ public int consumerMaxRate = ClientSessionFactoryImpl.DEFAULT_CONSUMER_MAX_RATE; - public int producerWindowSize = ClientSessionFactoryImpl.DEFAULT_PRODUCER_WINDOW_SIZE; + public int confirmationWindowSize = ClientSessionFactoryImpl.DEFAULT_CONFIRMATION_WINDOW_SIZE; public int producerMaxRate = ClientSessionFactoryImpl.DEFAULT_PRODUCER_MAX_RATE; @@ -269,14 +269,14 @@ this.consumerMaxRate = consumerMaxRate; } - public int getProducerWindowSize() + public int getConfirmationWindowSize() { - return producerWindowSize; + return confirmationWindowSize; } - public void setProducerWindowSize(int producerWindowSize) + public void setConfirmationWindowSize(int confirmationWindowSize) { - this.producerWindowSize = producerWindowSize; + this.confirmationWindowSize = confirmationWindowSize; } public int getProducerMaxRate() Modified: trunk/src/main/org/hornetq/jms/server/impl/JMSServerDeployer.java =================================================================== --- trunk/src/main/org/hornetq/jms/server/impl/JMSServerDeployer.java 2009-10-30 15:12:21 UTC (rev 8170) +++ trunk/src/main/org/hornetq/jms/server/impl/JMSServerDeployer.java 2009-10-30 17:41:19 UTC (rev 8171) @@ -132,7 +132,7 @@ int transactionBatchSize = getInteger(e, "transaction-batch-size", ClientSessionFactoryImpl.DEFAULT_ACK_BATCH_SIZE, GT_ZERO); int consumerWindowSize = getInteger(e, "consumer-window-size", ClientSessionFactoryImpl.DEFAULT_CONSUMER_WINDOW_SIZE, GE_ZERO); int consumerMaxRate = getInteger(e, "consumer-max-rate", ClientSessionFactoryImpl.DEFAULT_CONSUMER_MAX_RATE, MINUS_ONE_OR_GT_ZERO); - int producerWindowSize = getInteger(e, "producer-window-size", ClientSessionFactoryImpl.DEFAULT_PRODUCER_WINDOW_SIZE, GT_ZERO); + int confirmationWindowSize = getInteger(e, "confirmation-window-size", ClientSessionFactoryImpl.DEFAULT_CONFIRMATION_WINDOW_SIZE, MINUS_ONE_OR_GT_ZERO); int producerMaxRate = getInteger(e, "producer-max-rate", ClientSessionFactoryImpl.DEFAULT_PRODUCER_MAX_RATE, MINUS_ONE_OR_GT_ZERO); boolean cacheLargeMessagesClient = getBoolean(e, "cache-large-message-client", ClientSessionFactoryImpl.DEFAULT_CACHE_LARGE_MESSAGE_CLIENT); int minLargeMessageSize = getInteger(e, "min-large-message-size", ClientSessionFactoryImpl.DEFAULT_MIN_LARGE_MESSAGE_SIZE, GT_ZERO); @@ -240,7 +240,7 @@ minLargeMessageSize, consumerWindowSize, consumerMaxRate, - producerWindowSize, + confirmationWindowSize, producerMaxRate, blockOnAcknowledge, blockOnPersistentSend, @@ -273,7 +273,7 @@ minLargeMessageSize, consumerWindowSize, consumerMaxRate, - producerWindowSize, + confirmationWindowSize, producerMaxRate, blockOnAcknowledge, blockOnPersistentSend, Modified: trunk/src/main/org/hornetq/jms/server/impl/JMSServerManagerImpl.java =================================================================== --- trunk/src/main/org/hornetq/jms/server/impl/JMSServerManagerImpl.java 2009-10-30 15:12:21 UTC (rev 8170) +++ trunk/src/main/org/hornetq/jms/server/impl/JMSServerManagerImpl.java 2009-10-30 17:41:19 UTC (rev 8171) @@ -370,7 +370,7 @@ int minLargeMessageSize, int consumerWindowSize, int consumerMaxRate, - int producerWindowSize, + int confirmationWindowSize, int producerMaxRate, boolean blockOnAcknowledge, boolean blockOnPersistentSend, @@ -403,7 +403,7 @@ cf.setMinLargeMessageSize(minLargeMessageSize); cf.setConsumerWindowSize(consumerWindowSize); cf.setConsumerMaxRate(consumerMaxRate); - cf.setProducerWindowSize(producerWindowSize); + cf.setConfirmationWindowSize(confirmationWindowSize); cf.setProducerMaxRate(producerMaxRate); cf.setBlockOnAcknowledge(blockOnAcknowledge); cf.setBlockOnPersistentSend(blockOnPersistentSend); @@ -438,7 +438,7 @@ int minLargeMessageSize, int consumerWindowSize, int consumerMaxRate, - int producerWindowSize, + int confirmationWindowSize, int producerMaxRate, boolean blockOnAcknowledge, boolean blockOnPersistentSend, @@ -473,7 +473,7 @@ cf.setMinLargeMessageSize(minLargeMessageSize); cf.setConsumerWindowSize(consumerWindowSize); cf.setConsumerMaxRate(consumerMaxRate); - cf.setProducerWindowSize(producerWindowSize); + cf.setConfirmationWindowSize(confirmationWindowSize); cf.setProducerMaxRate(producerMaxRate); cf.setBlockOnAcknowledge(blockOnAcknowledge); cf.setBlockOnPersistentSend(blockOnPersistentSend); @@ -753,7 +753,7 @@ config.getMinLargeMessageSize(), config.getConsumerWindowSize(), config.getConsumerMaxRate(), - config.getProducerWindowSize(), + config.getConfirmationWindowSize(), config.getProducerMaxRate(), config.isBlockOnAcknowledge(), config.isBlockOnPersistentSend(), @@ -786,7 +786,7 @@ config.getMinLargeMessageSize(), config.getConsumerWindowSize(), config.getConsumerMaxRate(), - config.getProducerWindowSize(), + config.getConfirmationWindowSize(), config.getProducerMaxRate(), config.isBlockOnAcknowledge(), config.isBlockOnPersistentSend(), Modified: trunk/src/main/org/hornetq/jms/server/management/ConnectionFactoryControl.java =================================================================== --- trunk/src/main/org/hornetq/jms/server/management/ConnectionFactoryControl.java 2009-10-30 15:12:21 UTC (rev 8170) +++ trunk/src/main/org/hornetq/jms/server/management/ConnectionFactoryControl.java 2009-10-30 17:41:19 UTC (rev 8171) @@ -42,7 +42,7 @@ int getProducerMaxRate(); - int getProducerWindowSize(); + int getConfirmationWindowSize(); boolean isBlockOnAcknowledge(); Modified: trunk/src/main/org/hornetq/jms/server/management/JMSServerControl.java =================================================================== --- trunk/src/main/org/hornetq/jms/server/management/JMSServerControl.java 2009-10-30 15:12:21 UTC (rev 8170) +++ trunk/src/main/org/hornetq/jms/server/management/JMSServerControl.java 2009-10-30 17:41:19 UTC (rev 8171) @@ -103,7 +103,7 @@ int minLargeMessageSize, int consumerWindowSize, int consumerMaxRate, - int producerWindowSize, + int confirmationWindowSize, int producerMaxRate, boolean blockOnAcknowledge, boolean blockOnPersistentSend, @@ -136,7 +136,7 @@ @Parameter(name = "minLargeMessageSize") int minLargeMessageSize, @Parameter(name = "consumerWindowSize") int consumerWindowSize, @Parameter(name = "consumerMaxRate") int consumerMaxRate, - @Parameter(name = "producerWindowSize") int producerWindowSize, + @Parameter(name = "confirmationWindowSize") int confirmationWindowSize, @Parameter(name = "producerMaxRate") int producerMaxRate, @Parameter(name = "blockOnAcknowledge") boolean blockOnAcknowledge, @Parameter(name = "blockOnPersistentSend") boolean blockOnPersistentSend, @@ -181,7 +181,7 @@ int minLargeMessageSize, int consumerWindowSize, int consumerMaxRate, - int producerWindowSize, + int confirmationWindowSize, int producerMaxRate, boolean blockOnAcknowledge, boolean blockOnPersistentSend, @@ -215,7 +215,7 @@ @Parameter(name = "minLargeMessageSize") int minLargeMessageSize, @Parameter(name = "consumerWindowSize") int consumerWindowSize, @Parameter(name = "consumerMaxRate") int consumerMaxRate, - @Parameter(name = "producerWindowSize") int producerWindowSize, + @Parameter(name = "confirmationWindowSize") int confirmationWindowSize, @Parameter(name = "producerMaxRate") int producerMaxRate, @Parameter(name = "blockOnAcknowledge") boolean blockOnAcknowledge, @Parameter(name = "blockOnPersistentSend") boolean blockOnPersistentSend, Modified: trunk/src/main/org/hornetq/jms/server/management/impl/JMSConnectionFactoryControlImpl.java =================================================================== --- trunk/src/main/org/hornetq/jms/server/management/impl/JMSConnectionFactoryControlImpl.java 2009-10-30 15:12:21 UTC (rev 8170) +++ trunk/src/main/org/hornetq/jms/server/management/impl/JMSConnectionFactoryControlImpl.java 2009-10-30 17:41:19 UTC (rev 8171) @@ -90,9 +90,9 @@ return cf.getProducerMaxRate(); } - public int getProducerWindowSize() + public int getConfirmationWindowSize() { - return cf.getProducerWindowSize(); + return cf.getConfirmationWindowSize(); } public int getDupsOKBatchSize() Modified: trunk/src/main/org/hornetq/jms/server/management/impl/JMSServerControlImpl.java =================================================================== --- trunk/src/main/org/hornetq/jms/server/management/impl/JMSServerControlImpl.java 2009-10-30 15:12:21 UTC (rev 8170) +++ trunk/src/main/org/hornetq/jms/server/management/impl/JMSServerControlImpl.java 2009-10-30 17:41:19 UTC (rev 8171) @@ -230,7 +230,7 @@ final int minLargeMessageSize, final int consumerWindowSize, final int consumerMaxRate, - final int producerWindowSize, + final int confirmationWindowSize, final int producerMaxRate, final boolean blockOnAcknowledge, final boolean blockOnPersistentSend, @@ -267,7 +267,7 @@ minLargeMessageSize, consumerWindowSize, consumerMaxRate, - producerWindowSize, + confirmationWindowSize, producerMaxRate, blockOnAcknowledge, blockOnPersistentSend, @@ -303,7 +303,7 @@ final int minLargeMessageSize, final int consumerWindowSize, final int consumerMaxRate, - final int producerWindowSize, + final int confirmationWindowSize, final int producerMaxRate, final boolean blockOnAcknowledge, final boolean blockOnPersistentSend, @@ -342,7 +342,7 @@ minLargeMessageSize, consumerWindowSize, consumerMaxRate, - producerWindowSize, + confirmationWindowSize, producerMaxRate, blockOnAcknowledge, blockOnPersistentSend, @@ -399,7 +399,7 @@ final int minLargeMessageSize, final int consumerWindowSize, final int consumerMaxRate, - final int producerWindowSize, + final int confirmationWindowSize, final int producerMaxRate, final boolean blockOnAcknowledge, final boolean blockOnPersistentSend, @@ -434,7 +434,7 @@ minLargeMessageSize, consumerWindowSize, consumerMaxRate, - producerWindowSize, + confirmationWindowSize, producerMaxRate, blockOnAcknowledge, blockOnPersistentSend, @@ -470,7 +470,7 @@ final int minLargeMessageSize, final int consumerWindowSize, final int consumerMaxRate, - final int producerWindowSize, + final int confirmationWindowSize, final int producerMaxRate, final boolean blockOnAcknowledge, final boolean blockOnPersistentSend, @@ -505,7 +505,7 @@ minLargeMessageSize, consumerWindowSize, consumerMaxRate, - producerWindowSize, + confirmationWindowSize, producerMaxRate, blockOnAcknowledge, blockOnPersistentSend, Modified: trunk/src/main/org/hornetq/jms/server/recovery/HornetQXAResourceWrapper.java =================================================================== --- trunk/src/main/org/hornetq/jms/server/recovery/HornetQXAResourceWrapper.java 2009-10-30 15:12:21 UTC (rev 8170) +++ trunk/src/main/org/hornetq/jms/server/recovery/HornetQXAResourceWrapper.java 2009-10-30 17:41:19 UTC (rev 8171) @@ -21,11 +21,11 @@ import org.hornetq.core.client.ClientSession; import org.hornetq.core.client.ClientSessionFactory; +import org.hornetq.core.client.SessionFailureListener; import org.hornetq.core.client.impl.ClientSessionFactoryImpl; import org.hornetq.core.config.TransportConfiguration; import org.hornetq.core.exception.HornetQException; import org.hornetq.core.logging.Logger; -import org.hornetq.core.remoting.FailureListener; /** * XAResourceWrapper. @@ -42,7 +42,7 @@ * * @version $Revision: 45341 $ */ -public class HornetQXAResourceWrapper implements XAResource, FailureListener +public class HornetQXAResourceWrapper implements XAResource, SessionFailureListener { /** The log */ private static final Logger log = Logger.getLogger(HornetQXAResourceWrapper.class); @@ -216,6 +216,10 @@ log.warn("Notified of connection failure in recovery connectionFactory for provider " + connectorFactoryClassName, me); close(); } + + public void beforeReconnect(HornetQException me) + { + } /** * Get the connectionFactory XAResource Modified: trunk/src/main/org/hornetq/ra/ConnectionFactoryProperties.java =================================================================== --- trunk/src/main/org/hornetq/ra/ConnectionFactoryProperties.java 2009-10-30 15:12:21 UTC (rev 8170) +++ trunk/src/main/org/hornetq/ra/ConnectionFactoryProperties.java 2009-10-30 17:41:19 UTC (rev 8171) @@ -75,7 +75,7 @@ private Integer consumerMaxRate; - private Integer producerWindowSize; + private Integer confirmationWindowSize; private Integer producerMaxRate; @@ -413,24 +413,24 @@ this.consumerMaxRate = consumerMaxRate; } - public Integer getProducerWindowSize() + public Integer getConfirmationWindowSize() { if (trace) { - log.trace("getProducerWindowSize()"); + log.trace("getConfirmationWindowSize()"); } hasBeenUpdated = true; - return producerWindowSize; + return confirmationWindowSize; } - public void setProducerWindowSize(Integer producerWindowSize) + public void setConfirmationWindowSize(Integer confirmationWindowSize) { if (trace) { - log.trace("setProducerWindowSize(" + producerWindowSize + ")"); + log.trace("setConfirmationWindowSize(" + confirmationWindowSize + ")"); } hasBeenUpdated = true; - this.producerWindowSize = producerWindowSize; + this.confirmationWindowSize = confirmationWindowSize; } public Integer getProducerMaxRate() Modified: trunk/src/main/org/hornetq/ra/HornetQRAManagedConnectionFactory.java =================================================================== --- trunk/src/main/org/hornetq/ra/HornetQRAManagedConnectionFactory.java 2009-10-30 15:12:21 UTC (rev 8170) +++ trunk/src/main/org/hornetq/ra/HornetQRAManagedConnectionFactory.java 2009-10-30 17:41:19 UTC (rev 8171) @@ -541,14 +541,14 @@ mcfProperties.setConsumerMaxRate(consumerMaxRate); } - public Integer getProducerWindowSize() + public Integer getConfirmationWindowSize() { - return mcfProperties.getProducerWindowSize(); + return mcfProperties.getConfirmationWindowSize(); } - public void setProducerWindowSize(Integer producerWindowSize) + public void setConfirmationWindowSize(Integer confirmationWindowSize) { - mcfProperties.setProducerWindowSize(producerWindowSize); + mcfProperties.setConfirmationWindowSize(confirmationWindowSize); } public Integer getProducerMaxRate() Modified: trunk/src/main/org/hornetq/ra/HornetQResourceAdapter.java =================================================================== --- trunk/src/main/org/hornetq/ra/HornetQResourceAdapter.java 2009-10-30 15:12:21 UTC (rev 8170) +++ trunk/src/main/org/hornetq/ra/HornetQResourceAdapter.java 2009-10-30 17:41:19 UTC (rev 8171) @@ -628,33 +628,33 @@ } /** - * Get producer window size + * Get confirmation window size * * @return The value */ - public Integer getProducerWindowSize() + public Integer getConfirmationWindowSize() { if (trace) { - log.trace("getProducerWindowSize()"); + log.trace("getConfirmationWindowSize()"); } - return raProperties.getProducerWindowSize(); + return raProperties.getConfirmationWindowSize(); } /** - * Set producer window size + * Set confirmation window size * - * @param producerWindowSize The value + * @param confirmationWindowSize The value */ - public void setProducerWindowSize(final Integer producerWindowSize) + public void setConfirmationWindowSize(final Integer confirmationWindowSize) { if (trace) { - log.trace("setProducerWindowSize(" + producerWindowSize + ")"); + log.trace("setConfirmationWindowSize(" + confirmationWindowSize + ")"); } - raProperties.setProducerWindowSize(producerWindowSize); + raProperties.setConfirmationWindowSize(confirmationWindowSize); } /** @@ -1477,11 +1477,11 @@ { cf.setProducerMaxRate(val2); } - val2 = overrideProperties.getProducerWindowSize() != null ? overrideProperties.getProducerWindowSize() - : raProperties.getProducerWindowSize(); + val2 = overrideProperties.getConfirmationWindowSize() != null ? overrideProperties.getConfirmationWindowSize() + : raProperties.getConfirmationWindowSize(); if (val2 != null) { - cf.setProducerWindowSize(val2); + cf.setConfirmationWindowSize(val2); } val2 = overrideProperties.getReconnectAttempts() != null ? overrideProperties.getReconnectAttempts() : raProperties.getReconnectAttempts(); Modified: trunk/tests/config/hornetq-jms-for-JMSServerDeployerTest.xml =================================================================== --- trunk/tests/config/hornetq-jms-for-JMSServerDeployerTest.xml 2009-10-30 15:12:21 UTC (rev 8170) +++ trunk/tests/config/hornetq-jms-for-JMSServerDeployerTest.xml 2009-10-30 17:41:19 UTC (rev 8171) @@ -14,7 +14,7 @@ <call-timeout>5678</call-timeout> <consumer-window-size>12345</consumer-window-size> <consumer-max-rate>6789</consumer-max-rate> - <producer-window-size>123456</producer-window-size> + <confirmation-window-size>123456</confirmation-window-size> <producer-max-rate>789</producer-max-rate> <min-large-message-size>12</min-large-message-size> <client-id>TestClientID</client-id> Modified: trunk/tests/jms-tests/src/org/hornetq/jms/tests/BrowserTest.java =================================================================== --- trunk/tests/jms-tests/src/org/hornetq/jms/tests/BrowserTest.java 2009-10-30 15:12:21 UTC (rev 8170) +++ trunk/tests/jms-tests/src/org/hornetq/jms/tests/BrowserTest.java 2009-10-30 17:41:19 UTC (rev 8171) @@ -35,7 +35,6 @@ */ public class BrowserTest extends JMSTestCase { - // Constants ----------------------------------------------------------------------------------- // Static --------------------------------------------------------------------------------------- Modified: trunk/tests/jms-tests/src/org/hornetq/jms/tests/CTSMiscellaneousTest.java =================================================================== --- trunk/tests/jms-tests/src/org/hornetq/jms/tests/CTSMiscellaneousTest.java 2009-10-30 15:12:21 UTC (rev 8170) +++ trunk/tests/jms-tests/src/org/hornetq/jms/tests/CTSMiscellaneousTest.java 2009-10-30 17:41:19 UTC (rev 8171) @@ -27,7 +27,7 @@ import static org.hornetq.core.client.impl.ClientSessionFactoryImpl.DEFAULT_MIN_LARGE_MESSAGE_SIZE; import static org.hornetq.core.client.impl.ClientSessionFactoryImpl.DEFAULT_PRE_ACKNOWLEDGE; import static org.hornetq.core.client.impl.ClientSessionFactoryImpl.DEFAULT_PRODUCER_MAX_RATE; -import static org.hornetq.core.client.impl.ClientSessionFactoryImpl.DEFAULT_PRODUCER_WINDOW_SIZE; +import static org.hornetq.core.client.impl.ClientSessionFactoryImpl.DEFAULT_CONFIRMATION_WINDOW_SIZE; import static org.hornetq.core.client.impl.ClientSessionFactoryImpl.DEFAULT_RECONNECT_ATTEMPTS; import static org.hornetq.core.client.impl.ClientSessionFactoryImpl.DEFAULT_RETRY_INTERVAL; import static org.hornetq.core.client.impl.ClientSessionFactoryImpl.DEFAULT_RETRY_INTERVAL_MULTIPLIER; @@ -96,7 +96,7 @@ DEFAULT_MIN_LARGE_MESSAGE_SIZE, DEFAULT_CONSUMER_WINDOW_SIZE, DEFAULT_CONSUMER_MAX_RATE, - DEFAULT_PRODUCER_WINDOW_SIZE, + DEFAULT_CONFIRMATION_WINDOW_SIZE, DEFAULT_PRODUCER_MAX_RATE, true, true, Modified: trunk/tests/jms-tests/src/org/hornetq/jms/tests/JMSTestCase.java =================================================================== --- trunk/tests/jms-tests/src/org/hornetq/jms/tests/JMSTestCase.java 2009-10-30 15:12:21 UTC (rev 8170) +++ trunk/tests/jms-tests/src/org/hornetq/jms/tests/JMSTestCase.java 2009-10-30 17:41:19 UTC (rev 8171) @@ -27,7 +27,7 @@ import static org.hornetq.core.client.impl.ClientSessionFactoryImpl.DEFAULT_MIN_LARGE_MESSAGE_SIZE; import static org.hornetq.core.client.impl.ClientSessionFactoryImpl.DEFAULT_PRE_ACKNOWLEDGE; import static org.hornetq.core.client.impl.ClientSessionFactoryImpl.DEFAULT_PRODUCER_MAX_RATE; -import static org.hornetq.core.client.impl.ClientSessionFactoryImpl.DEFAULT_PRODUCER_WINDOW_SIZE; +import static org.hornetq.core.client.impl.ClientSessionFactoryImpl.DEFAULT_CONFIRMATION_WINDOW_SIZE; import static org.hornetq.core.client.impl.ClientSessionFactoryImpl.DEFAULT_RECONNECT_ATTEMPTS; import static org.hornetq.core.client.impl.ClientSessionFactoryImpl.DEFAULT_RETRY_INTERVAL; import static org.hornetq.core.client.impl.ClientSessionFactoryImpl.DEFAULT_RETRY_INTERVAL_MULTIPLIER; @@ -92,7 +92,7 @@ DEFAULT_MIN_LARGE_MESSAGE_SIZE, DEFAULT_CONSUMER_WINDOW_SIZE, DEFAULT_CONSUMER_MAX_RATE, - DEFAULT_PRODUCER_WINDOW_SIZE, + DEFAULT_CONFIRMATION_WINDOW_SIZE, DEFAULT_PRODUCER_MAX_RATE, true, true, 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 2009-10-30 15:12:21 UTC (rev 8170) +++ trunk/tests/jms-tests/src/org/hornetq/jms/tests/message/MessageHeaderTest.java 2009-10-30 17:41:19 UTC (rev 8171) @@ -39,6 +39,7 @@ import org.hornetq.core.client.ClientProducer; import org.hornetq.core.client.ClientSession; import org.hornetq.core.client.SendAcknowledgementHandler; +import org.hornetq.core.client.SessionFailureListener; import org.hornetq.core.client.impl.ClientMessageImpl; import org.hornetq.core.exception.HornetQException; import org.hornetq.core.remoting.FailureListener; @@ -1156,6 +1157,45 @@ public void start(Xid xid, int i) throws XAException { } + + /* (non-Javadoc) + * @see org.hornetq.core.client.ClientSession#createBuffer(byte[]) + */ + public HornetQBuffer createBuffer(byte[] bytes) + { + // TODO Auto-generated method stub + return null; + } + + /* (non-Javadoc) + * @see org.hornetq.core.client.ClientSession#createBuffer(int) + */ + public HornetQBuffer createBuffer(int size) + { + // TODO Auto-generated method stub + return null; + } + + /* (non-Javadoc) + * @see org.hornetq.core.client.ClientSession#createClientMessage(boolean, org.hornetq.core.remoting.spi.HornetQBuffer) + */ + public ClientMessage createClientMessage(boolean durable, HornetQBuffer buffer) + { + // TODO Auto-generated method stub + return null; + } + + public void addFailureListener(SessionFailureListener listener) + { + // TODO Auto-generated method stub + + } + + public boolean removeFailureListener(SessionFailureListener listener) + { + // TODO Auto-generated method stub + return false; + } } } Modified: trunk/tests/jms-tests/src/org/hornetq/jms/tests/tools/container/LocalTestServer.java =================================================================== --- trunk/tests/jms-tests/src/org/hornetq/jms/tests/tools/container/LocalTestServer.java 2009-10-30 15:12:21 UTC (rev 8170) +++ trunk/tests/jms-tests/src/org/hornetq/jms/tests/tools/container/LocalTestServer.java 2009-10-30 17:41:19 UTC (rev 8171) @@ -26,7 +26,7 @@ import static org.hornetq.core.client.impl.ClientSessionFactoryImpl.DEFAULT_MIN_LARGE_MESSAGE_SIZE; import static org.hornetq.core.client.impl.ClientSessionFactoryImpl.DEFAULT_PRE_ACKNOWLEDGE; import static org.hornetq.core.client.impl.ClientSessionFactoryImpl.DEFAULT_PRODUCER_MAX_RATE; -import static org.hornetq.core.client.impl.ClientSessionFactoryImpl.DEFAULT_PRODUCER_WINDOW_SIZE; +import static org.hornetq.core.client.impl.ClientSessionFactoryImpl.DEFAULT_CONFIRMATION_WINDOW_SIZE; import static org.hornetq.core.client.impl.ClientSessionFactoryImpl.DEFAULT_RECONNECT_ATTEMPTS; import static org.hornetq.core.client.impl.ClientSessionFactoryImpl.DEFAULT_RETRY_INTERVAL; import static org.hornetq.core.client.impl.ClientSessionFactoryImpl.DEFAULT_RETRY_INTERVAL_MULTIPLIER; @@ -321,7 +321,7 @@ DEFAULT_MIN_LARGE_MESSAGE_SIZE, prefetchSize, DEFAULT_CONSUMER_MAX_RATE, - DEFAULT_PRODUCER_WINDOW_SIZE, + DEFAULT_CONFIRMATION_WINDOW_SIZE, DEFAULT_PRODUCER_MAX_RATE, blockOnAcknowledge, true, Added: trunk/tests/src/org/hornetq/tests/integration/EncodeSizeTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/EncodeSizeTest.java (rev 0) +++ trunk/tests/src/org/hornetq/tests/integration/EncodeSizeTest.java 2009-10-30 17:41:19 UTC (rev 8171) @@ -0,0 +1,137 @@ +/* + * 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; + +import org.hornetq.core.buffers.ChannelBuffers; +import org.hornetq.core.client.ClientMessage; +import org.hornetq.core.client.impl.ClientMessageImpl; +import org.hornetq.core.logging.Logger; +import org.hornetq.core.remoting.impl.wireformat.SessionSendLargeMessage; +import org.hornetq.core.remoting.impl.wireformat.SessionSendMessage; +import org.hornetq.core.remoting.spi.HornetQBuffer; +import org.hornetq.core.server.ServerMessage; +import org.hornetq.core.server.impl.ServerMessageImpl; +import org.hornetq.tests.util.RandomUtil; +import org.hornetq.tests.util.UnitTestCase; + +/** + * A EncodeSizeTest + * + * For flow control, it's crucial that encode sizes on client and server are the same + * + * @author Tim Fox + * + * + */ +public class EncodeSizeTest extends UnitTestCase +{ + private static final Logger log = Logger.getLogger(EncodeSizeTest.class); + + public void testMessageEncodeSize() throws Exception + { + for (int i = 0; i < 10; i++) + { + ClientMessage clientMessage = new ClientMessageImpl(0); + + clientMessage.putIntProperty(RandomUtil.randomString(), RandomUtil.randomInt()); + clientMessage.putBooleanProperty(RandomUtil.randomString(), RandomUtil.randomBoolean()); + clientMessage.putByteProperty(RandomUtil.randomString(), RandomUtil.randomByte()); + clientMessage.putBytesProperty(RandomUtil.randomString(), RandomUtil.randomBytes(125)); + clientMessage.putDoubleProperty(RandomUtil.randomString(), RandomUtil.randomDouble()); + clientMessage.putFloatProperty(RandomUtil.randomString(), RandomUtil.randomFloat()); + clientMessage.putLongProperty(RandomUtil.randomString(), RandomUtil.randomLong()); + clientMessage.putShortProperty(RandomUtil.randomString(), RandomUtil.randomShort()); + clientMessage.putStringProperty(RandomUtil.randomString(), RandomUtil.randomString()); + + clientMessage.setDestination(RandomUtil.randomSimpleString()); + + byte[] bytes = RandomUtil.randomBytes(1000); + + HornetQBuffer body = ChannelBuffers.dynamicBuffer(bytes); + + clientMessage.setBody(body); + + int clientEncodeSize = clientMessage.getEncodeSize(); + + HornetQBuffer buffer = ChannelBuffers.dynamicBuffer(clientEncodeSize); + + clientMessage.encode(buffer); + + int wireSize = buffer.writerIndex(); + + assertEquals(clientEncodeSize, wireSize); + + ServerMessage serverMessage = new ServerMessageImpl(); + + serverMessage.decode(buffer); + + int serverEncodeSize = serverMessage.getEncodeSize(); + + assertEquals(wireSize, serverEncodeSize); + } + } + + public void testMessageEncodeSizeWithPacket() throws Exception + { + for (int i = 0; i < 10; i++) + { + ClientMessage clientMessage = new ClientMessageImpl(0); + + clientMessage.putIntProperty(RandomUtil.randomString(), RandomUtil.randomInt()); + clientMessage.putBooleanProperty(RandomUtil.randomString(), RandomUtil.randomBoolean()); + clientMessage.putByteProperty(RandomUtil.randomString(), RandomUtil.randomByte()); + clientMessage.putBytesProperty(RandomUtil.randomString(), RandomUtil.randomBytes(125)); + clientMessage.putDoubleProperty(RandomUtil.randomString(), RandomUtil.randomDouble()); + clientMessage.putFloatProperty(RandomUtil.randomString(), RandomUtil.randomFloat()); + clientMessage.putLongProperty(RandomUtil.randomString(), RandomUtil.randomLong()); + clientMessage.putShortProperty(RandomUtil.randomString(), RandomUtil.randomShort()); + clientMessage.putStringProperty(RandomUtil.randomString(), RandomUtil.randomString()); + + clientMessage.setDestination(RandomUtil.randomSimpleString()); + + byte[] bytes = RandomUtil.randomBytes(1000); + + HornetQBuffer body = ChannelBuffers.dynamicBuffer(bytes); + + clientMessage.setBody(body); + + int clientEncodeSize = clientMessage.getEncodeSize(); + + SessionSendMessage packet = new SessionSendMessage(clientMessage, false); + + HornetQBuffer buffer = ChannelBuffers.dynamicBuffer(packet.getRequiredBufferSize()); + + packet.encode(buffer); + + int wireSize = buffer.writerIndex(); + + assertEquals(wireSize, packet.getRequiredBufferSize()); + + SessionSendMessage received = new SessionSendMessage(); + + //The length + buffer.readInt(); + //The packet type byte + buffer.readByte(); + + received.decode(buffer); + + ServerMessage serverMessage = received.getServerMessage(); + + int serverEncodeSize = serverMessage.getEncodeSize(); + + assertEquals(clientEncodeSize, serverEncodeSize); + } + } +} Modified: trunk/tests/src/org/hornetq/tests/integration/client/AddressSettingsTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/client/AddressSettingsTest.java 2009-10-30 15:12:21 UTC (rev 8170) +++ trunk/tests/src/org/hornetq/tests/integration/client/AddressSettingsTest.java 2009-10-30 17:41:19 UTC (rev 8171) @@ -37,8 +37,6 @@ private SimpleString addressC = new SimpleString("addressC"); - private SimpleString addressC2 = new SimpleString("add.addressC"); - private SimpleString queueA = new SimpleString("queueA"); private SimpleString queueB = new SimpleString("queueB"); Modified: trunk/tests/src/org/hornetq/tests/integration/client/ConsumerTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/client/ConsumerTest.java 2009-10-30 15:12:21 UTC (rev 8170) +++ trunk/tests/src/org/hornetq/tests/integration/client/ConsumerTest.java 2009-10-30 17:41:19 UTC (rev 8171) @@ -15,8 +15,6 @@ import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; -import javax.jms.ObjectMessage; - import org.hornetq.core.client.ClientConsumer; import org.hornetq.core.client.ClientMessage; import org.hornetq.core.client.ClientProducer; @@ -244,7 +242,7 @@ } }); ClientSessionFactory sfReceive = createInVMFactory(); - sfReceive.setProducerWindowSize(100); + sfReceive.setConfirmationWindowSize(100); sfReceive.setAckBatchSize(-1); ClientSession sessionRec = sfReceive.createSession(false, true, true); ClientConsumer consumer = sessionRec.createConsumer(QUEUE); Modified: trunk/tests/src/org/hornetq/tests/integration/client/HornetQCrashTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/client/HornetQCrashTest.java 2009-10-30 15:12:21 UTC (rev 8170) +++ trunk/tests/src/org/hornetq/tests/integration/client/HornetQCrashTest.java 2009-10-30 17:41:19 UTC (rev 8171) @@ -52,7 +52,7 @@ ClientSessionFactory clientSessionFactory = new ClientSessionFactoryImpl(new TransportConfiguration(InVMConnectorFactory.class.getName())); // Force an ack at once - this means the send call will block - clientSessionFactory.setProducerWindowSize(1); + clientSessionFactory.setConfirmationWindowSize(1); ClientSession session = clientSessionFactory.createSession(); Modified: trunk/tests/src/org/hornetq/tests/integration/client/LargeMessageTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/client/LargeMessageTest.java 2009-10-30 15:12:21 UTC (rev 8170) +++ trunk/tests/src/org/hornetq/tests/integration/client/LargeMessageTest.java 2009-10-30 17:41:19 UTC (rev 8171) @@ -80,8 +80,6 @@ server.start(); - log.info("*********** starting test"); - ClientSessionFactory sf = createInVMFactory(); session = sf.createSession(false, false, false); @@ -92,8 +90,6 @@ Message clientFile = createLargeClientMessage(session, messageSize, true); - log.info("*********** sending large message"); - producer.send(clientFile); session.commit(); @@ -153,8 +149,6 @@ server.start(); - log.info("*********** starting test"); - ClientSessionFactory sf = createInVMFactory(); session = sf.createSession(false, false, false); @@ -177,8 +171,6 @@ Message clientFile = createLargeClientMessage(session, messageSize, true); - log.info("*********** sending large message"); - producer.send(clientFile); session.commit(); @@ -1124,7 +1116,6 @@ public void testPageOnLargeMessageNullPersistence() throws Exception { testPageOnLargeMessage(false, false); - } public void testSendSmallMessageXA() throws Exception @@ -1328,8 +1319,6 @@ server.stop(); - log.info("Restartning"); - server = createServer(true); server.start(); @@ -1484,8 +1473,6 @@ session.start(); - log.info("Session started"); - ClientProducer producer = session.createProducer(ADDRESS); ClientConsumer consumer = session.createConsumer(ADDRESS); @@ -2028,7 +2015,7 @@ producer.send(message); } - + ClientMessage clientFile = createLargeClientMessage(session, numberOfBytesBigMessage); producer.send(clientFile); @@ -2055,8 +2042,6 @@ { ClientMessage message2 = consumer.receive(RECEIVE_WAIT_TIME); - log.debug("got message " + i); - assertNotNull(message2); message2.acknowledge(); Modified: trunk/tests/src/org/hornetq/tests/integration/client/PagingTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/client/PagingTest.java 2009-10-30 15:12:21 UTC (rev 8170) +++ trunk/tests/src/org/hornetq/tests/integration/client/PagingTest.java 2009-10-30 17:41:19 UTC (rev 8171) @@ -30,6 +30,7 @@ import org.hornetq.core.remoting.spi.HornetQBuffer; import org.hornetq.core.server.HornetQServer; import org.hornetq.core.server.Queue; +import org.hornetq.core.settings.impl.AddressFullMessagePolicy; import org.hornetq.core.settings.impl.AddressSettings; import org.hornetq.tests.util.ServiceTestBase; import org.hornetq.utils.DataConstants; @@ -635,7 +636,7 @@ HashMap<String, AddressSettings> settings = new HashMap<String, AddressSettings>(); AddressSettings set = new AddressSettings(); - set.setDropMessagesWhenFull(true); + set.setAddressFullMessagePolicy(AddressFullMessagePolicy.DROP); settings.put(ADDRESS.toString(), set); Added: trunk/tests/src/org/hornetq/tests/integration/client/ProducerFlowControlTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/client/ProducerFlowControlTest.java (rev 0) +++ trunk/tests/src/org/hornetq/tests/integration/client/ProducerFlowControlTest.java 2009-10-30 17:41:19 UTC (rev 8171) @@ -0,0 +1,719 @@ +/* + * 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.concurrent.CountDownLatch; +import java.util.concurrent.atomic.AtomicBoolean; + +import org.hornetq.core.client.ClientConsumer; +import org.hornetq.core.client.ClientMessage; +import org.hornetq.core.client.ClientProducer; +import org.hornetq.core.client.ClientSession; +import org.hornetq.core.client.ClientSessionFactory; +import org.hornetq.core.client.MessageHandler; +import org.hornetq.core.client.impl.ClientMessageImpl; +import org.hornetq.core.config.Configuration; +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; +import org.hornetq.tests.util.ServiceTestBase; +import org.hornetq.utils.SimpleString; + +/** + * + * A ProducerFlowControlTest + * + * @author tim fox + * + */ +public class ProducerFlowControlTest extends ServiceTestBase +{ + private static final Logger log = Logger.getLogger(ProducerFlowControlTest.class); + + // TODO need to test crashing a producer with unused credits returns them to the pool + + public void testFlowControlSingleConsumer() throws Exception + { + testFlowControl(false, 1000, 500, 10 * 1024, 1024, 1024, 1024, 1, 1, 0, false); + } + + public void testFlowControlAnon() throws Exception + { + testFlowControl(false, 1000, 500, 10 * 1024, 1024, 1024, 1024, 1, 1, 0, true); + } + + public void testFlowControlSingleConsumerLargeMaxSize() throws Exception + { + testFlowControl(false, 1000, 500, 1024 * 1024, 1024, 1024, 1024, 1, 1, 0, false); + } + + public void testFlowControlMultipleConsumers() throws Exception + { + testFlowControl(false, 1000, 500, 10 * 1024, 1024, 1024, 1024, 5, 1, 0, false); + } + + public void testFlowControlZeroConsumerWindowSize() throws Exception + { + testFlowControl(false, 1000, 500, 10 * 1024, 1024, 0, 1024, 1, 1, 0, false); + } + + public void testFlowControlZeroAckBatchSize() throws Exception + { + testFlowControl(false, 1000, 500, 10 * 1024, 1024, 1024, 0, 1, 1, 0, false); + } + + public void testFlowControlSingleConsumerSlowConsumer() throws Exception + { + testFlowControl(false, 100, 500, 1024, 512, 512, 512, 1, 1, 10, false); + } + + public void testFlowControlSmallMessages() throws Exception + { + testFlowControl(false, 1000, 0, 10 * 1024, 1024, 1024, 1024, 1, 1, 0, false); + } + + public void testFlowControlLargeMessagesSmallWindowSize() throws Exception + { + testFlowControl(false, 1000, 10 * 1024, 10 * 1024, 1024, 1024, 1024, 1, 1, 0, false); + } + + public void testFlowControlMultipleProducers() throws Exception + { + testFlowControl(false, 1000, 500, 1024 * 1024, 1024, 1024, 1024, 1, 5, 0, false); + } + + public void testFlowControlMultipleProducersAndConsumers() throws Exception + { + testFlowControl(false, 500, 500, 100 * 1024, 1024, 1024, 1024, 1, 3, 3, false); + } + + public void testFlowControlMultipleProducersAnon() throws Exception + { + testFlowControl(false, 1000, 500, 1024 * 1024, 1024, 1024, 1024, 1, 5, 0, true); + } + + public void testFlowControlSingleConsumerNetty() throws Exception + { + testFlowControl(true, 1000, 500, 10 * 1024, 1024, 1024, 1024, 1, 1, 0, false); + } + + public void testFlowControlSingleConsumerLargeMaxSizeNetty() throws Exception + { + testFlowControl(true, 1000, 500, 1024 * 1024, 1024, 1024, 1024, 1, 1, 0, false); + } + + public void testFlowControlMultipleConsumersNetty() throws Exception + { + testFlowControl(true, 1000, 500, 10 * 1024, 1024, 1024, 1024, 5, 1, 0, false); + } + + public void testFlowControlZeroConsumerWindowSizeNetty() throws Exception + { + testFlowControl(true, 1000, 500, 10 * 1024, 1024, 0, 1024, 1, 1, 0, false); + } + + public void testFlowControlZeroAckBatchSizeNetty() throws Exception + { + testFlowControl(true, 1000, 500, 10 * 1024, 1024, 1024, 0, 1, 1, 0, false); + } + + public void testFlowControlSingleConsumerSlowConsumerNetty() throws Exception + { + testFlowControl(true, 100, 500, 1024, 512, 512, 512, 1, 1, 10, false); + } + + public void testFlowControlSmallMessagesNetty() throws Exception + { + testFlowControl(true, 1000, 0, 10 * 1024, 1024, 1024, 1024, 1, 1, 0, false); + } + + public void testFlowControlLargeMessagesSmallWindowSizeNetty() throws Exception + { + testFlowControl(true, 1000, 10 * 1024, 10 * 1024, 1024, 1024, 1024, 1, 1, 0, false); + } + + public void testFlowControlMultipleProducersNetty() throws Exception + { + testFlowControl(true, 1000, 500, 1024 * 1024, 1024, 1024, 1024, 1, 5, 0, false); + } + + public void testFlowControlMultipleProducersAndConsumersNetty() throws Exception + { + testFlowControl(false, 500, 500, 100 * 1024, 1024, 1024, 1024, 1, 3, 3, false); + } + + public void testFlowControlMultipleProducersAnonNetty() throws Exception + { + testFlowControl(true, 1000, 500, 1024 * 1024, 1024, 1024, 1024, 1, 5, 0, true); + } + + private void testFlowControl(final boolean netty, + final int numMessages, + final int messageSize, + final int maxSize, + final int producerWindowSize, + final int consumerWindowSize, + final int ackBatchSize, + final int numConsumers, + final int numProducers, + final long consumerDelay, + final boolean anon) throws Exception + { + final SimpleString address = new SimpleString("testaddress"); + + Configuration config = super.createDefaultConfig(netty); + + HornetQServer server = createServer(false, config); + + AddressSettings addressSettings = new AddressSettings(); + addressSettings.setMaxSizeBytes(maxSize); + addressSettings.setAddressFullMessagePolicy(AddressFullMessagePolicy.BLOCK); + + HierarchicalRepository<AddressSettings> repos = server.getAddressSettingsRepository(); + repos.addMatch(address.toString(), addressSettings); + + server.start(); + + ClientSessionFactory sf; + + if (netty) + { + sf = createNettyFactory(); + } + else + { + sf = createInVMFactory(); + } + + sf.setProducerWindowSize(producerWindowSize); + sf.setConsumerWindowSize(consumerWindowSize); + sf.setAckBatchSize(ackBatchSize); + + ClientSession session = sf.createSession(false, true, true, true); + + final String queueName = "testqueue"; + + for (int i = 0; i < numConsumers; i++) + { + session.createQueue(address, new SimpleString(queueName + i), null, false); + } + + session.start(); + + class MyHandler implements MessageHandler + { + int count = 0; + + final CountDownLatch latch = new CountDownLatch(1); + + volatile Exception exception; + + public void onMessage(ClientMessage message) + { + try + { + message.acknowledge(); + + if (++count == numMessages * numProducers) + { + latch.countDown(); + } + + if (consumerDelay > 0) + { + Thread.sleep(consumerDelay); + } + } + catch (Exception e) + { + log.error("Failed to handle message", e); + + this.exception = e; + } + } + } + + MyHandler[] handlers = new MyHandler[numConsumers]; + + for (int i = 0; i < numConsumers; i++) + { + handlers[i] = new MyHandler(); + + ClientConsumer consumer = session.createConsumer(new SimpleString(queueName + i)); + + consumer.setMessageHandler(handlers[i]); + } + + ClientProducer[] producers = new ClientProducer[numProducers]; + + for (int i = 0; i < numProducers; i++) + { + if (anon) + { + producers[i] = session.createProducer(); + } + else + { + producers[i] = session.createProducer(address); + } + } + + long start = System.currentTimeMillis(); + + byte[] bytes = new byte[messageSize]; + + for (int i = 0; i < numMessages; i++) + { + ClientMessage message = new ClientMessageImpl(false, bytes); + + for (int j = 0; j < numProducers; j++) + { + if (anon) + { + producers[j].send(address, message); + } + else + { + producers[j].send(message); + } + + } + } + + for (int i = 0; i < numConsumers; i++) + { + handlers[i].latch.await(); + + assertNull(handlers[i].exception); + } + + long end = System.currentTimeMillis(); + + double rate = 1000 * (double)numMessages / (end - start); + + log.info("rate is " + rate + " msgs / sec"); + + session.close(); + + TestSupportPageStore store = (TestSupportPageStore)server.getPostOffice() + .getPagingManager() + .getPageStore(address); + + assertFalse(store.isExceededAvailableCredits()); + + server.stop(); + } + + public void testUnusedCreditsAreReturnedOnSessionCloseWithWaitingEntries() throws Exception + { + final SimpleString address = new SimpleString("testaddress"); + + Configuration config = super.createDefaultConfig(false); + + HornetQServer server = createServer(false, config); + + AddressSettings addressSettings = new AddressSettings(); + addressSettings.setMaxSizeBytes(1024); + addressSettings.setAddressFullMessagePolicy(AddressFullMessagePolicy.BLOCK); + + HierarchicalRepository<AddressSettings> repos = server.getAddressSettingsRepository(); + repos.addMatch(address.toString(), addressSettings); + + server.start(); + + ClientSessionFactory sf = createInVMFactory(); + + sf.setProducerWindowSize(1024); + 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); + + assertEquals(1, waiting); + + byte[] bytes = new byte[0]; + + ClientMessage message = new ClientMessageImpl(false, 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 = new ClientMessageImpl(false, bytes); + + producer2.send(message2); + + // Make sure it blocked until the first producer was closed + assertTrue(closer.closed); + + t.join(); + + session2.close(); + + TestSupportPageStore store = (TestSupportPageStore)server.getPostOffice() + .getPagingManager() + .getPageStore(address); + + assertFalse(store.isExceededAvailableCredits()); + + server.stop(); + } + + public void testUnusedCreditsAreReturnedOnSessionCloseNoWaitingEntries() throws Exception + { + final SimpleString address = new SimpleString("testaddress"); + + Configuration config = super.createDefaultConfig(false); + + HornetQServer server = createServer(false, config); + + AddressSettings addressSettings = new AddressSettings(); + addressSettings.setMaxSizeBytes(1024); + addressSettings.setAddressFullMessagePolicy(AddressFullMessagePolicy.BLOCK); + + HierarchicalRepository<AddressSettings> repos = server.getAddressSettingsRepository(); + repos.addMatch(address.toString(), addressSettings); + + server.start(); + + ClientSessionFactory sf = createInVMFactory(); + + sf.setProducerWindowSize(1024); + 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 = new ClientMessageImpl(false, 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); + + assertEquals(1, waiting); + + message = new ClientMessageImpl(false, bytes); + + producer2.send(message); + + session2.close(); + + TestSupportPageStore store = (TestSupportPageStore)server.getPostOffice() + .getPagingManager() + .getPageStore(address); + + assertFalse(store.isExceededAvailableCredits()); + + server.stop(); + } + + public void testUnusedCreditsAreReturnedWaitingEntriesForClosedSessions() throws Exception + { + final SimpleString address = new SimpleString("testaddress"); + + Configuration config = super.createDefaultConfig(false); + + HornetQServer server = createServer(false, config); + + AddressSettings addressSettings = new AddressSettings(); + addressSettings.setMaxSizeBytes(1024); + addressSettings.setAddressFullMessagePolicy(AddressFullMessagePolicy.BLOCK); + + HierarchicalRepository<AddressSettings> repos = server.getAddressSettingsRepository(); + repos.addMatch(address.toString(), addressSettings); + + server.start(); + + ClientSessionFactory sf = createInVMFactory(); + + sf.setProducerWindowSize(1024); + 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 = new ClientMessageImpl(false, 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); + + 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); + + assertEquals(2, waiting); + + session2.close(); + + session.close(); + + message = new ClientMessageImpl(false, bytes); + + producer3.send(message); + + session3.close(); + + TestSupportPageStore store = (TestSupportPageStore)server.getPostOffice() + .getPagingManager() + .getPageStore(address); + + assertFalse(store.isExceededAvailableCredits()); + + server.stop(); + } + + public void testClosingSessionUnblocksBlockedProducer() throws Exception + { + final SimpleString address = new SimpleString("testaddress"); + + Configuration config = super.createDefaultConfig(false); + + HornetQServer server = createServer(false, config); + + AddressSettings addressSettings = new AddressSettings(); + addressSettings.setMaxSizeBytes(1024); + addressSettings.setAddressFullMessagePolicy(AddressFullMessagePolicy.BLOCK); + + HierarchicalRepository<AddressSettings> repos = server.getAddressSettingsRepository(); + repos.addMatch(address.toString(), addressSettings); + + server.start(); + + ClientSessionFactory sf = createInVMFactory(); + + sf.setProducerWindowSize(1024); + 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[2000]; + + ClientMessage message = new ClientMessageImpl(false, bytes); + + final AtomicBoolean closed = new AtomicBoolean(false); + + Thread t = new Thread(new Runnable() + { + public void run() + { + try + { + Thread.sleep(500); + + closed.set(true); + + session.close(); + } + catch (Exception e) + { + } + } + }); + + t.start(); + + // This will block + producer.send(message); + + assertTrue(closed.get()); + + t.join(); + + TestSupportPageStore store = (TestSupportPageStore)server.getPostOffice() + .getPagingManager() + .getPageStore(address); + + assertFalse(store.isExceededAvailableCredits()); + + server.stop(); + } + + public void testFlowControlMessageNotRouted() throws Exception + { + final SimpleString address = new SimpleString("testaddress"); + + Configuration config = super.createDefaultConfig(false); + + HornetQServer server = createServer(false, config); + + AddressSettings addressSettings = new AddressSettings(); + addressSettings.setMaxSizeBytes(1024); + addressSettings.setAddressFullMessagePolicy(AddressFullMessagePolicy.BLOCK); + + HierarchicalRepository<AddressSettings> repos = server.getAddressSettingsRepository(); + repos.addMatch(address.toString(), addressSettings); + + server.start(); + + ClientSessionFactory sf = createInVMFactory(); + + sf.setProducerWindowSize(1024); + sf.setConsumerWindowSize(1024); + sf.setAckBatchSize(1024); + + final ClientSession session = sf.createSession(false, true, true, true); + + ClientProducer producer = session.createProducer(address); + + byte[] bytes = new byte[100]; + + final int numMessages = 1000; + + for (int i = 0; i < numMessages; i++) + { + ClientMessage message = new ClientMessageImpl(false, bytes); + + producer.send(message); + } + + session.close(); + + TestSupportPageStore store = (TestSupportPageStore)server.getPostOffice() + .getPagingManager() + .getPageStore(address); + + assertFalse(store.isExceededAvailableCredits()); + + server.stop(); + } + + +} Modified: trunk/tests/src/org/hornetq/tests/integration/client/ProducerTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/client/ProducerTest.java 2009-10-30 15:12:21 UTC (rev 8170) +++ trunk/tests/src/org/hornetq/tests/integration/client/ProducerTest.java 2009-10-30 17:41:19 UTC (rev 8171) @@ -75,7 +75,7 @@ } }); ClientSessionFactory cf = createInVMFactory(); - cf.setProducerWindowSize(100); + cf.setConfirmationWindowSize(100); ClientSession session = cf.createSession(false, true, true); ClientProducer producer = session.createProducer(QUEUE); ClientMessage message = session.createClientMessage(true); Deleted: trunk/tests/src/org/hornetq/tests/integration/client/SendTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/client/SendTest.java 2009-10-30 15:12:21 UTC (rev 8170) +++ trunk/tests/src/org/hornetq/tests/integration/client/SendTest.java 2009-10-30 17:41:19 UTC (rev 8171) @@ -1,113 +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.integration.client; - -import org.hornetq.core.client.ClientProducer; -import org.hornetq.core.client.ClientSession; -import org.hornetq.core.client.ClientSessionFactory; -import org.hornetq.core.server.HornetQServer; -import org.hornetq.core.server.Queue; -import org.hornetq.tests.util.ServiceTestBase; -import org.hornetq.utils.SimpleString; - -/** - * @author <a href="mailto:andy.taylor@jboss.org">Andy Taylor</a> - */ -public class SendTest extends ServiceTestBase -{ - public final SimpleString addressA = new SimpleString("addressA"); - - public final SimpleString queueA = new SimpleString("queueA"); - - public final SimpleString queueB = new SimpleString("queueB"); - - public final SimpleString queueC = new SimpleString("queueC"); - - - public void testSendWithCommit() throws Exception - { - HornetQServer server = createServer(false); - try - { - server.start(); - ClientSessionFactory cf = createInVMFactory(); - ClientSession session = cf.createSession(false, false, false); - session.createQueue(addressA, queueA, false); - ClientProducer cp = session.createProducer(addressA); - int numMessages = 100; - for (int i = 0; i < numMessages; i++) - { - cp.send(session.createClientMessage(false)); - } - Queue q = (Queue) server.getPostOffice().getBinding(queueA).getBindable(); - assertEquals(q.getMessageCount(), 0); - session.commit(); - assertEquals(q.getMessageCount(), numMessages); - // now send some more - for (int i = 0; i < numMessages; i++) - { - cp.send(session.createClientMessage(false)); - } - assertEquals(q.getMessageCount(), numMessages); - session.commit(); - assertEquals(q.getMessageCount(), numMessages * 2); - session.close(); - } - finally - { - if (server.isStarted()) - { - server.stop(); - } - } - } - - public void testSendWithRollback() throws Exception - { - HornetQServer server = createServer(false); - try - { - server.start(); - ClientSessionFactory cf = createInVMFactory(); - ClientSession session = cf.createSession(false, false, false); - session.createQueue(addressA, queueA, false); - ClientProducer cp = session.createProducer(addressA); - int numMessages = 100; - for (int i = 0; i < numMessages; i++) - { - cp.send(session.createClientMessage(false)); - } - Queue q = (Queue) server.getPostOffice().getBinding(queueA).getBindable(); - assertEquals(q.getMessageCount(), 0); - session.rollback(); - assertEquals(q.getMessageCount(), 0); - // now send some more - for (int i = 0; i < numMessages; i++) - { - cp.send(session.createClientMessage(false)); - } - assertEquals(q.getMessageCount(), 0); - session.commit(); - assertEquals(q.getMessageCount(), numMessages); - session.close(); - } - finally - { - if (server.isStarted()) - { - server.stop(); - } - } - } - -} Modified: trunk/tests/src/org/hornetq/tests/integration/client/SessionFactoryTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/client/SessionFactoryTest.java 2009-10-30 15:12:21 UTC (rev 8170) +++ trunk/tests/src/org/hornetq/tests/integration/client/SessionFactoryTest.java 2009-10-30 17:41:19 UTC (rev 8171) @@ -122,7 +122,7 @@ ClientSessionFactoryImpl.DEFAULT_MIN_LARGE_MESSAGE_SIZE, ClientSessionFactoryImpl.DEFAULT_CONSUMER_WINDOW_SIZE, ClientSessionFactoryImpl.DEFAULT_CONSUMER_MAX_RATE, - ClientSessionFactoryImpl.DEFAULT_PRODUCER_WINDOW_SIZE, + ClientSessionFactoryImpl.DEFAULT_CONFIRMATION_WINDOW_SIZE, ClientSessionFactoryImpl.DEFAULT_PRODUCER_MAX_RATE, ClientSessionFactoryImpl.DEFAULT_BLOCK_ON_ACKNOWLEDGE, ClientSessionFactoryImpl.DEFAULT_BLOCK_ON_PERSISTENT_SEND, @@ -182,7 +182,7 @@ ClientSessionFactoryImpl.DEFAULT_MIN_LARGE_MESSAGE_SIZE, ClientSessionFactoryImpl.DEFAULT_CONSUMER_WINDOW_SIZE, ClientSessionFactoryImpl.DEFAULT_CONSUMER_MAX_RATE, - ClientSessionFactoryImpl.DEFAULT_PRODUCER_WINDOW_SIZE, + ClientSessionFactoryImpl.DEFAULT_CONFIRMATION_WINDOW_SIZE, ClientSessionFactoryImpl.DEFAULT_PRODUCER_MAX_RATE, ClientSessionFactoryImpl.DEFAULT_BLOCK_ON_ACKNOWLEDGE, ClientSessionFactoryImpl.DEFAULT_BLOCK_ON_PERSISTENT_SEND, @@ -232,7 +232,7 @@ ClientSessionFactoryImpl.DEFAULT_MIN_LARGE_MESSAGE_SIZE, ClientSessionFactoryImpl.DEFAULT_CONSUMER_WINDOW_SIZE, ClientSessionFactoryImpl.DEFAULT_CONSUMER_MAX_RATE, - ClientSessionFactoryImpl.DEFAULT_PRODUCER_WINDOW_SIZE, + ClientSessionFactoryImpl.DEFAULT_CONFIRMATION_WINDOW_SIZE, ClientSessionFactoryImpl.DEFAULT_PRODUCER_MAX_RATE, ClientSessionFactoryImpl.DEFAULT_BLOCK_ON_ACKNOWLEDGE, ClientSessionFactoryImpl.DEFAULT_BLOCK_ON_PERSISTENT_SEND, @@ -282,7 +282,7 @@ ClientSessionFactoryImpl.DEFAULT_MIN_LARGE_MESSAGE_SIZE, ClientSessionFactoryImpl.DEFAULT_CONSUMER_WINDOW_SIZE, ClientSessionFactoryImpl.DEFAULT_CONSUMER_MAX_RATE, - ClientSessionFactoryImpl.DEFAULT_PRODUCER_WINDOW_SIZE, + ClientSessionFactoryImpl.DEFAULT_CONFIRMATION_WINDOW_SIZE, ClientSessionFactoryImpl.DEFAULT_PRODUCER_MAX_RATE, ClientSessionFactoryImpl.DEFAULT_BLOCK_ON_ACKNOWLEDGE, ClientSessionFactoryImpl.DEFAULT_BLOCK_ON_PERSISTENT_SEND, @@ -332,7 +332,7 @@ ClientSessionFactoryImpl.DEFAULT_MIN_LARGE_MESSAGE_SIZE, ClientSessionFactoryImpl.DEFAULT_CONSUMER_WINDOW_SIZE, ClientSessionFactoryImpl.DEFAULT_CONSUMER_MAX_RATE, - ClientSessionFactoryImpl.DEFAULT_PRODUCER_WINDOW_SIZE, + ClientSessionFactoryImpl.DEFAULT_CONFIRMATION_WINDOW_SIZE, ClientSessionFactoryImpl.DEFAULT_PRODUCER_MAX_RATE, ClientSessionFactoryImpl.DEFAULT_BLOCK_ON_ACKNOWLEDGE, ClientSessionFactoryImpl.DEFAULT_BLOCK_ON_PERSISTENT_SEND, @@ -378,7 +378,7 @@ int minLargeMessageSize = RandomUtil.randomPositiveInt(); int consumerWindowSize = RandomUtil.randomPositiveInt(); int consumerMaxRate = RandomUtil.randomPositiveInt(); - int producerWindowSize = RandomUtil.randomPositiveInt(); + int confirmationWindowSize = RandomUtil.randomPositiveInt(); int producerMaxRate = RandomUtil.randomPositiveInt(); boolean blockOnAcknowledge = RandomUtil.randomBoolean(); boolean blockOnPersistentSend = RandomUtil.randomBoolean(); @@ -406,7 +406,7 @@ cf.setMinLargeMessageSize(minLargeMessageSize); cf.setConsumerWindowSize(consumerWindowSize); cf.setConsumerMaxRate(consumerMaxRate); - cf.setProducerWindowSize(producerWindowSize); + cf.setConfirmationWindowSize(confirmationWindowSize); cf.setProducerMaxRate(producerMaxRate); cf.setBlockOnAcknowledge(blockOnAcknowledge); cf.setBlockOnPersistentSend(blockOnPersistentSend); @@ -434,7 +434,7 @@ assertEquals(minLargeMessageSize, cf.getMinLargeMessageSize()); assertEquals(consumerWindowSize, cf.getConsumerWindowSize()); assertEquals(consumerMaxRate, cf.getConsumerMaxRate()); - assertEquals(producerWindowSize, cf.getProducerWindowSize()); + assertEquals(confirmationWindowSize, cf.getConfirmationWindowSize()); assertEquals(producerMaxRate, cf.getProducerMaxRate()); assertEquals(blockOnAcknowledge, cf.isBlockOnAcknowledge()); assertEquals(blockOnPersistentSend, cf.isBlockOnPersistentSend()); @@ -470,7 +470,7 @@ int minLargeMessageSize = RandomUtil.randomPositiveInt(); int consumerWindowSize = RandomUtil.randomPositiveInt(); int consumerMaxRate = RandomUtil.randomPositiveInt(); - int producerWindowSize = RandomUtil.randomPositiveInt(); + int confirmationWindowSize = RandomUtil.randomPositiveInt(); int producerMaxRate = RandomUtil.randomPositiveInt(); boolean blockOnAcknowledge = RandomUtil.randomBoolean(); boolean blockOnPersistentSend = RandomUtil.randomBoolean(); @@ -580,7 +580,7 @@ } try { - cf.setProducerWindowSize(producerWindowSize); + cf.setConfirmationWindowSize(confirmationWindowSize); fail("Should throw exception"); } catch (IllegalStateException e) @@ -742,7 +742,7 @@ cf.getMinLargeMessageSize(); cf.getConsumerWindowSize(); cf.getConsumerMaxRate(); - cf.getProducerWindowSize(); + cf.getConfirmationWindowSize(); cf.getProducerMaxRate(); cf.isBlockOnAcknowledge(); cf.isBlockOnPersistentSend(); @@ -773,7 +773,7 @@ int minLargeMessageSize, int consumerWindowSize, int consumerMaxRate, - int producerWindowSize, + int confirmationWindowSize, int producerMaxRate, boolean blockOnAcknowledge, boolean blockOnPersistentSend, @@ -814,7 +814,7 @@ assertEquals(cf.getMinLargeMessageSize(), minLargeMessageSize); assertEquals(cf.getConsumerWindowSize(), consumerWindowSize); assertEquals(cf.getConsumerMaxRate(), consumerMaxRate); - assertEquals(cf.getProducerWindowSize(), producerWindowSize); + assertEquals(cf.getConfirmationWindowSize(), confirmationWindowSize); assertEquals(cf.getProducerMaxRate(), producerMaxRate); assertEquals(cf.isBlockOnAcknowledge(), blockOnAcknowledge); assertEquals(cf.isBlockOnPersistentSend(), blockOnPersistentSend); Modified: trunk/tests/src/org/hornetq/tests/integration/client/SessionSendAcknowledgementHandlerTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/client/SessionSendAcknowledgementHandlerTest.java 2009-10-30 15:12:21 UTC (rev 8170) +++ trunk/tests/src/org/hornetq/tests/integration/client/SessionSendAcknowledgementHandlerTest.java 2009-10-30 17:41:19 UTC (rev 8171) @@ -72,7 +72,7 @@ { ClientSessionFactory csf = createInVMFactory(); - csf.setProducerWindowSize(1024); + csf.setConfirmationWindowSize(1024); ClientSession session = csf.createSession(null, null, false, true, true, false, 1); Modified: trunk/tests/src/org/hornetq/tests/integration/client/SessionTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/client/SessionTest.java 2009-10-30 15:12:21 UTC (rev 8170) +++ trunk/tests/src/org/hornetq/tests/integration/client/SessionTest.java 2009-10-30 17:41:19 UTC (rev 8171) @@ -21,9 +21,9 @@ import org.hornetq.core.client.ClientProducer; import org.hornetq.core.client.ClientSession; import org.hornetq.core.client.ClientSessionFactory; +import org.hornetq.core.client.SessionFailureListener; import org.hornetq.core.client.impl.ClientSessionInternal; import org.hornetq.core.exception.HornetQException; -import org.hornetq.core.remoting.FailureListener; import org.hornetq.core.remoting.RemotingConnection; import org.hornetq.core.remoting.impl.wireformat.SessionBindingQueryResponseMessage; import org.hornetq.core.remoting.impl.wireformat.SessionQueueQueryResponseMessage; @@ -50,12 +50,16 @@ ClientSessionFactory cf = createInVMFactory(); ClientSession clientSession = cf.createSession(false, true, true); final CountDownLatch latch = new CountDownLatch(1); - clientSession.addFailureListener(new FailureListener() + clientSession.addFailureListener(new SessionFailureListener() { public void connectionFailed(HornetQException me) { latch.countDown(); } + + public void beforeReconnect(HornetQException me) + { + } }); //Make sure failure listener is called if server is stopped without session being closed first @@ -83,7 +87,7 @@ server.start(); ClientSessionFactory cf = createInVMFactory(); ClientSession clientSession = cf.createSession(false, true, true); - class MyFailureListener implements FailureListener + class MyFailureListener implements SessionFailureListener { boolean called = false; @@ -91,6 +95,10 @@ { called = true; } + + public void beforeReconnect(HornetQException me) + { + } } MyFailureListener listener = new MyFailureListener(); Copied: trunk/tests/src/org/hornetq/tests/integration/client/TransactionalSendTest.java (from rev 8139, trunk/tests/src/org/hornetq/tests/integration/client/SendTest.java) =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/client/TransactionalSendTest.java (rev 0) +++ trunk/tests/src/org/hornetq/tests/integration/client/TransactionalSendTest.java 2009-10-30 17:41:19 UTC (rev 8171) @@ -0,0 +1,113 @@ +/* + * 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 org.hornetq.core.client.ClientProducer; +import org.hornetq.core.client.ClientSession; +import org.hornetq.core.client.ClientSessionFactory; +import org.hornetq.core.server.HornetQServer; +import org.hornetq.core.server.Queue; +import org.hornetq.tests.util.ServiceTestBase; +import org.hornetq.utils.SimpleString; + +/** + * @author <a href="mailto:andy.taylor@jboss.org">Andy Taylor</a> + */ +public class TransactionalSendTest extends ServiceTestBase +{ + public final SimpleString addressA = new SimpleString("addressA"); + + public final SimpleString queueA = new SimpleString("queueA"); + + public final SimpleString queueB = new SimpleString("queueB"); + + public final SimpleString queueC = new SimpleString("queueC"); + + + public void testSendWithCommit() throws Exception + { + HornetQServer server = createServer(false); + try + { + server.start(); + ClientSessionFactory cf = createInVMFactory(); + ClientSession session = cf.createSession(false, false, false); + session.createQueue(addressA, queueA, false); + ClientProducer cp = session.createProducer(addressA); + int numMessages = 100; + for (int i = 0; i < numMessages; i++) + { + cp.send(session.createClientMessage(false)); + } + Queue q = (Queue) server.getPostOffice().getBinding(queueA).getBindable(); + assertEquals(q.getMessageCount(), 0); + session.commit(); + assertEquals(q.getMessageCount(), numMessages); + // now send some more + for (int i = 0; i < numMessages; i++) + { + cp.send(session.createClientMessage(false)); + } + assertEquals(q.getMessageCount(), numMessages); + session.commit(); + assertEquals(q.getMessageCount(), numMessages * 2); + session.close(); + } + finally + { + if (server.isStarted()) + { + server.stop(); + } + } + } + + public void testSendWithRollback() throws Exception + { + HornetQServer server = createServer(false); + try + { + server.start(); + ClientSessionFactory cf = createInVMFactory(); + ClientSession session = cf.createSession(false, false, false); + session.createQueue(addressA, queueA, false); + ClientProducer cp = session.createProducer(addressA); + int numMessages = 100; + for (int i = 0; i < numMessages; i++) + { + cp.send(session.createClientMessage(false)); + } + Queue q = (Queue) server.getPostOffice().getBinding(queueA).getBindable(); + assertEquals(q.getMessageCount(), 0); + session.rollback(); + assertEquals(q.getMessageCount(), 0); + // now send some more + for (int i = 0; i < numMessages; i++) + { + cp.send(session.createClientMessage(false)); + } + assertEquals(q.getMessageCount(), 0); + session.commit(); + assertEquals(q.getMessageCount(), numMessages); + session.close(); + } + finally + { + if (server.isStarted()) + { + server.stop(); + } + } + } + +} Modified: trunk/tests/src/org/hornetq/tests/integration/cluster/bridge/BridgeReconnectTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/cluster/bridge/BridgeReconnectTest.java 2009-10-30 15:12:21 UTC (rev 8170) +++ trunk/tests/src/org/hornetq/tests/integration/cluster/bridge/BridgeReconnectTest.java 2009-10-30 17:41:19 UTC (rev 8171) @@ -489,8 +489,11 @@ ClientProducer prod0 = session0.createProducer(testAddress); + log.info("stopping server1"); server1.stop(); + log.info("restarting server1"); server1.start(); + log.info("server 1 restarted"); ClientSessionFactory csf1 = new ClientSessionFactoryImpl(server1tc); ClientSession session1 = csf1.createSession(false, true, true); @@ -510,13 +513,18 @@ prod0.send(message); } + + log.info("sent messages"); for (int i = 0; i < numMessages; i++) { - ClientMessage r1 = cons1.receive(1500); + ClientMessage r1 = cons1.receive(30000); assertNotNull(r1); assertEquals(i, r1.getProperty(propKey)); + log.info("got message " + r1.getProperty(propKey)); } + + log.info("got messages"); session0.close(); session1.close(); Modified: trunk/tests/src/org/hornetq/tests/integration/cluster/bridge/BridgeStartTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/cluster/bridge/BridgeStartTest.java 2009-10-30 15:12:21 UTC (rev 8170) +++ trunk/tests/src/org/hornetq/tests/integration/cluster/bridge/BridgeStartTest.java 2009-10-30 17:41:19 UTC (rev 8171) @@ -323,9 +323,9 @@ sf1.close(); log.info("stopping server 1"); - + server1.stop(); - + log.info("stopped server 1"); for (int i = 0; i < numMessages; i++) @@ -336,11 +336,11 @@ producer0.send(message); } - + log.info("sent some more messages"); server1.start(); - + log.info("started server1"); sf1 = new ClientSessionFactoryImpl(server1tc); @@ -350,7 +350,7 @@ consumer1 = session1.createConsumer(queueName1); session1.start(); - + log.info("started session"); for (int i = 0; i < numMessages; i++) @@ -546,7 +546,7 @@ 1d, 1, false, - false, + true, connectorPair); List<BridgeConfiguration> bridgeConfigs = new ArrayList<BridgeConfiguration>(); @@ -611,9 +611,11 @@ Bridge bridge = server0.getClusterManager().getBridges().get(bridgeName); + log.info("stopping bridge manually"); + bridge.stop(); - for (int i = 0; i < numMessages; i++) + for (int i = numMessages; i < numMessages * 2; i++) { ClientMessage message = session0.createClientMessage(false); @@ -626,7 +628,12 @@ bridge.start(); - for (int i = 0; i < numMessages; i++) + log.info("started bridge"); + + //The previous messages will get resent, but with duplicate detection they will be rejected + //at the target + + for (int i = numMessages; i < numMessages * 2; i++) { ClientMessage message = consumer1.receive(1000); @@ -679,7 +686,7 @@ server1.stop(); } - + protected void setUp() throws Exception { super.setUp(); Modified: trunk/tests/src/org/hornetq/tests/integration/cluster/distribution/ClusteredGroupingTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/cluster/distribution/ClusteredGroupingTest.java 2009-10-30 15:12:21 UTC (rev 8170) +++ trunk/tests/src/org/hornetq/tests/integration/cluster/distribution/ClusteredGroupingTest.java 2009-10-30 17:41:19 UTC (rev 8171) @@ -998,8 +998,6 @@ } - - public boolean isNetty() { return true; Modified: trunk/tests/src/org/hornetq/tests/integration/cluster/failover/AsynchronousFailoverTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/cluster/failover/AsynchronousFailoverTest.java 2009-10-30 15:12:21 UTC (rev 8170) +++ trunk/tests/src/org/hornetq/tests/integration/cluster/failover/AsynchronousFailoverTest.java 2009-10-30 17:41:19 UTC (rev 8171) @@ -22,13 +22,13 @@ import org.hornetq.core.client.ClientMessage; import org.hornetq.core.client.ClientProducer; import org.hornetq.core.client.ClientSession; +import org.hornetq.core.client.SessionFailureListener; import org.hornetq.core.client.impl.ClientSessionFactoryInternal; import org.hornetq.core.client.impl.ClientSessionInternal; import org.hornetq.core.client.impl.DelegatingSession; import org.hornetq.core.config.TransportConfiguration; import org.hornetq.core.exception.HornetQException; import org.hornetq.core.logging.Logger; -import org.hornetq.core.remoting.FailureListener; import org.hornetq.core.remoting.RemotingConnection; /** @@ -48,7 +48,7 @@ private volatile ClientSessionFactoryInternal sf; - class MyListener implements FailureListener + class MyListener implements SessionFailureListener { CountDownLatch latch = new CountDownLatch(1); @@ -56,6 +56,10 @@ { latch.countDown(); } + + public void beforeReconnect(HornetQException me) + { + } } public void testNonTransactional() throws Exception Modified: trunk/tests/src/org/hornetq/tests/integration/cluster/failover/FailoverTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/cluster/failover/FailoverTest.java 2009-10-30 15:12:21 UTC (rev 8170) +++ trunk/tests/src/org/hornetq/tests/integration/cluster/failover/FailoverTest.java 2009-10-30 17:41:19 UTC (rev 8171) @@ -29,6 +29,7 @@ import org.hornetq.core.client.ClientProducer; import org.hornetq.core.client.ClientSession; import org.hornetq.core.client.ClientSessionFactory; +import org.hornetq.core.client.SessionFailureListener; import org.hornetq.core.client.impl.ClientSessionFactoryImpl; import org.hornetq.core.client.impl.ClientSessionFactoryInternal; import org.hornetq.core.client.impl.ClientSessionInternal; @@ -36,7 +37,6 @@ import org.hornetq.core.exception.HornetQException; import org.hornetq.core.logging.Logger; import org.hornetq.core.message.impl.MessageImpl; -import org.hornetq.core.remoting.FailureListener; import org.hornetq.core.remoting.Interceptor; import org.hornetq.core.remoting.RemotingConnection; import org.hornetq.core.remoting.impl.invm.TransportConstants; @@ -89,6 +89,13 @@ public FailoverTest() { } + + abstract class BaseListener implements SessionFailureListener + { + public void beforeReconnect(HornetQException me) + { + } + } public void testNonTransacted() throws Exception { @@ -103,12 +110,12 @@ final CountDownLatch latch = new CountDownLatch(1); - class MyListener implements FailureListener + class MyListener extends BaseListener { public void connectionFailed(HornetQException me) { latch.countDown(); - } + } } session.addFailureListener(new MyListener()); @@ -195,12 +202,14 @@ final CountDownLatch latch = new CountDownLatch(1); - class MyListener implements FailureListener + class MyListener extends BaseListener { public void connectionFailed(HornetQException me) { latch.countDown(); } + + } session.addFailureListener(new MyListener()); @@ -263,7 +272,7 @@ final CountDownLatch latch = new CountDownLatch(1); - class MyListener implements FailureListener + class MyListener extends BaseListener { public void connectionFailed(HornetQException me) { @@ -344,7 +353,7 @@ final CountDownLatch latch = new CountDownLatch(1); - class MyListener implements FailureListener + class MyListener extends BaseListener { public void connectionFailed(HornetQException me) { @@ -433,7 +442,7 @@ final CountDownLatch latch = new CountDownLatch(1); - class MyListener implements FailureListener + class MyListener extends BaseListener { public void connectionFailed(HornetQException me) { @@ -516,7 +525,7 @@ final CountDownLatch latch = new CountDownLatch(1); - class MyListener implements FailureListener + class MyListener extends BaseListener { public void connectionFailed(HornetQException me) { @@ -613,7 +622,7 @@ final CountDownLatch latch = new CountDownLatch(1); - class MyListener implements FailureListener + class MyListener extends BaseListener { public void connectionFailed(HornetQException me) { @@ -683,7 +692,7 @@ final CountDownLatch latch = new CountDownLatch(1); - class MyListener implements FailureListener + class MyListener extends BaseListener { public void connectionFailed(HornetQException me) { @@ -756,7 +765,7 @@ final CountDownLatch latch = new CountDownLatch(1); - class MyListener implements FailureListener + class MyListener extends BaseListener { public void connectionFailed(HornetQException me) { @@ -830,7 +839,7 @@ final CountDownLatch latch = new CountDownLatch(1); - class MyListener implements FailureListener + class MyListener extends BaseListener { public void connectionFailed(HornetQException me) { @@ -917,7 +926,7 @@ final CountDownLatch latch = new CountDownLatch(1); - class MyListener implements FailureListener + class MyListener extends BaseListener { public void connectionFailed(HornetQException me) { @@ -1002,7 +1011,7 @@ final CountDownLatch latch = new CountDownLatch(1); - class MyListener implements FailureListener + class MyListener extends BaseListener { public void connectionFailed(HornetQException me) { @@ -1104,7 +1113,7 @@ final CountDownLatch latch = new CountDownLatch(1); - class MyListener implements FailureListener + class MyListener extends BaseListener { public void connectionFailed(HornetQException me) { @@ -1188,7 +1197,7 @@ final CountDownLatch latch = new CountDownLatch(1); - class MyListener implements FailureListener + class MyListener extends BaseListener { public void connectionFailed(HornetQException me) { @@ -1235,7 +1244,7 @@ Map<ClientSession, List<ClientConsumer>> sessionConsumerMap = new HashMap<ClientSession, List<ClientConsumer>>(); - class MyListener implements FailureListener + class MyListener extends BaseListener { CountDownLatch latch = new CountDownLatch(1); @@ -1350,7 +1359,7 @@ final CountDownLatch latch = new CountDownLatch(1); - class MyListener implements FailureListener + class MyListener extends BaseListener { public void connectionFailed(HornetQException me) { @@ -1430,7 +1439,7 @@ final CountDownLatch latch = new CountDownLatch(1); - class MyListener implements FailureListener + class MyListener extends BaseListener { public void connectionFailed(HornetQException me) { @@ -1513,7 +1522,7 @@ final CountDownLatch latch = new CountDownLatch(1); - class MyListener implements FailureListener + class MyListener extends BaseListener { public void connectionFailed(HornetQException me) { @@ -1610,7 +1619,7 @@ final CountDownLatch latch = new CountDownLatch(1); - class MyListener implements FailureListener + class MyListener extends BaseListener { public void connectionFailed(HornetQException me) { @@ -1678,7 +1687,7 @@ final CountDownLatch latch = new CountDownLatch(1); - class MyListener implements FailureListener + class MyListener extends BaseListener { public void connectionFailed(HornetQException me) { @@ -1834,7 +1843,7 @@ final CountDownLatch latch = new CountDownLatch(1); - class MyListener implements FailureListener + class MyListener extends BaseListener { public void connectionFailed(HornetQException me) { Modified: trunk/tests/src/org/hornetq/tests/integration/cluster/failover/GroupingFailoverSharedServerTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/cluster/failover/GroupingFailoverSharedServerTest.java 2009-10-30 15:12:21 UTC (rev 8170) +++ trunk/tests/src/org/hornetq/tests/integration/cluster/failover/GroupingFailoverSharedServerTest.java 2009-10-30 17:41:19 UTC (rev 8171) @@ -31,55 +31,54 @@ public class GroupingFailoverSharedServerTest extends GroupingFailoverTestBase { protected void setupReplicatedServer(int node, boolean fileStorage, boolean netty, int backupNode) + { + if (servers[node] != null) { - if (servers[node] != null) - { - throw new IllegalArgumentException("Already a server at node " + node); - } + throw new IllegalArgumentException("Already a server at node " + node); + } - Configuration configuration = new ConfigurationImpl(); + Configuration configuration = new ConfigurationImpl(); - configuration.setSecurityEnabled(false); - configuration.setBindingsDirectory(getBindingsDir(backupNode, false)); - configuration.setJournalMinFiles(2); - configuration.setJournalMaxAIO(1000); - configuration.setJournalDirectory(getJournalDir(backupNode, false)); - configuration.setJournalFileSize(100 * 1024); - configuration.setJournalType(JournalType.ASYNCIO); - configuration.setJournalMaxAIO(1000); - configuration.setPagingDirectory(getPageDir(backupNode, false)); - configuration.setLargeMessagesDirectory(getLargeMessagesDir(backupNode, false)); - configuration.setClustered(true); - configuration.setJournalCompactMinFiles(0); - configuration.setBackup(true); - configuration.setSharedStore(true); + configuration.setSecurityEnabled(false); + configuration.setBindingsDirectory(getBindingsDir(backupNode, false)); + configuration.setJournalMinFiles(2); + configuration.setJournalMaxAIO(1000); + configuration.setJournalDirectory(getJournalDir(backupNode, false)); + configuration.setJournalFileSize(100 * 1024); + configuration.setJournalType(JournalType.ASYNCIO); + configuration.setJournalMaxAIO(1000); + configuration.setPagingDirectory(getPageDir(backupNode, false)); + configuration.setLargeMessagesDirectory(getLargeMessagesDir(backupNode, false)); + configuration.setClustered(true); + configuration.setJournalCompactMinFiles(0); + configuration.setBackup(true); + configuration.setSharedStore(true); + configuration.getAcceptorConfigurations().clear(); - configuration.getAcceptorConfigurations().clear(); + Map<String, Object> params = generateParams(node, netty); - Map<String, Object> params = generateParams(node, netty); + TransportConfiguration invmtc = new TransportConfiguration(INVM_ACCEPTOR_FACTORY, params); + configuration.getAcceptorConfigurations().add(invmtc); - TransportConfiguration invmtc = new TransportConfiguration(INVM_ACCEPTOR_FACTORY, params); - configuration.getAcceptorConfigurations().add(invmtc); + if (netty) + { + TransportConfiguration nettytc = new TransportConfiguration(NETTY_ACCEPTOR_FACTORY, params); + configuration.getAcceptorConfigurations().add(nettytc); + } - if (netty) - { - TransportConfiguration nettytc = new TransportConfiguration(NETTY_ACCEPTOR_FACTORY, params); - configuration.getAcceptorConfigurations().add(nettytc); - } + HornetQServer server; - HornetQServer server; - - if (fileStorage) - { - server = HornetQ.newHornetQServer(configuration); - } - else - { - server = HornetQ.newHornetQServer(configuration, false); - } - servers[node] = server; + if (fileStorage) + { + server = HornetQ.newHornetQServer(configuration); } + else + { + server = HornetQ.newHornetQServer(configuration, false); + } + servers[node] = server; + } public void setupMasterServer(int i, boolean fileStorage, boolean netty) { Modified: trunk/tests/src/org/hornetq/tests/integration/cluster/failover/PagingFailoverTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/cluster/failover/PagingFailoverTest.java 2009-10-30 15:12:21 UTC (rev 8170) +++ trunk/tests/src/org/hornetq/tests/integration/cluster/failover/PagingFailoverTest.java 2009-10-30 17:41:19 UTC (rev 8171) @@ -22,12 +22,12 @@ import org.hornetq.core.client.ClientMessage; import org.hornetq.core.client.ClientProducer; import org.hornetq.core.client.ClientSession; +import org.hornetq.core.client.SessionFailureListener; import org.hornetq.core.client.impl.ClientSessionFactoryInternal; import org.hornetq.core.client.impl.ClientSessionInternal; import org.hornetq.core.config.Configuration; import org.hornetq.core.config.TransportConfiguration; import org.hornetq.core.exception.HornetQException; -import org.hornetq.core.remoting.FailureListener; import org.hornetq.core.remoting.RemotingConnection; import org.hornetq.core.server.HornetQServer; import org.hornetq.core.settings.impl.AddressSettings; @@ -88,12 +88,17 @@ final CountDownLatch latch = new CountDownLatch(1); - class MyListener implements FailureListener + class MyListener implements SessionFailureListener { public void connectionFailed(HornetQException me) { latch.countDown(); } + + public void beforeReconnect(HornetQException exception) + { + } + } session.addFailureListener(new MyListener()); Modified: trunk/tests/src/org/hornetq/tests/integration/cluster/reattach/MultiThreadRandomReattachTestBase.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/cluster/reattach/MultiThreadRandomReattachTestBase.java 2009-10-30 15:12:21 UTC (rev 8170) +++ trunk/tests/src/org/hornetq/tests/integration/cluster/reattach/MultiThreadRandomReattachTestBase.java 2009-10-30 17:41:19 UTC (rev 8171) @@ -1323,6 +1323,7 @@ { final ClientSessionFactoryInternal sf = new ClientSessionFactoryImpl(new TransportConfiguration("org.hornetq.core.remoting.impl.invm.InVMConnectorFactory")); sf.setReconnectAttempts(-1); + sf.setConfirmationWindowSize(1024 * 1024); return sf; } Modified: trunk/tests/src/org/hornetq/tests/integration/cluster/reattach/NettyMultiThreadRandomReattachTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/cluster/reattach/NettyMultiThreadRandomReattachTest.java 2009-10-30 15:12:21 UTC (rev 8170) +++ trunk/tests/src/org/hornetq/tests/integration/cluster/reattach/NettyMultiThreadRandomReattachTest.java 2009-10-30 17:41:19 UTC (rev 8171) @@ -49,6 +49,7 @@ { final ClientSessionFactoryInternal sf = new ClientSessionFactoryImpl(new TransportConfiguration("org.hornetq.integration.transports.netty.NettyConnectorFactory")); sf.setReconnectAttempts(-1); + sf.setConfirmationWindowSize(1024 * 1024); return sf; } Modified: trunk/tests/src/org/hornetq/tests/integration/cluster/reattach/RandomReattachTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/cluster/reattach/RandomReattachTest.java 2009-10-30 15:12:21 UTC (rev 8170) +++ trunk/tests/src/org/hornetq/tests/integration/cluster/reattach/RandomReattachTest.java 2009-10-30 17:41:19 UTC (rev 8171) @@ -221,6 +221,7 @@ ClientSessionFactoryImpl sf = new ClientSessionFactoryImpl(new TransportConfiguration("org.hornetq.core.remoting.impl.invm.InVMConnectorFactory")); sf.setReconnectAttempts(-1); + sf.setConfirmationWindowSize(1024 * 1024); ClientSession session = sf.createSession(false, false, false); Modified: trunk/tests/src/org/hornetq/tests/integration/cluster/reattach/ReattachTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/cluster/reattach/ReattachTest.java 2009-10-30 15:12:21 UTC (rev 8170) +++ trunk/tests/src/org/hornetq/tests/integration/cluster/reattach/ReattachTest.java 2009-10-30 17:41:19 UTC (rev 8171) @@ -17,6 +17,7 @@ import org.hornetq.core.client.ClientMessage; import org.hornetq.core.client.ClientProducer; import org.hornetq.core.client.ClientSession; +import org.hornetq.core.client.SessionFailureListener; import org.hornetq.core.client.impl.ClientSessionFactoryImpl; import org.hornetq.core.client.impl.ClientSessionFactoryInternal; import org.hornetq.core.client.impl.ClientSessionInternal; @@ -25,7 +26,6 @@ import org.hornetq.core.config.impl.ConfigurationImpl; import org.hornetq.core.exception.HornetQException; import org.hornetq.core.logging.Logger; -import org.hornetq.core.remoting.FailureListener; import org.hornetq.core.remoting.RemotingConnection; import org.hornetq.core.remoting.impl.invm.InVMConnector; import org.hornetq.core.remoting.impl.invm.InVMRegistry; @@ -79,6 +79,7 @@ sf.setRetryInterval(retryInterval); sf.setRetryIntervalMultiplier(retryMultiplier); sf.setReconnectAttempts(reconnectAttempts); + sf.setConfirmationWindowSize(1024 * 1024); ClientSession session = sf.createSession(false, true, true); @@ -156,6 +157,7 @@ sf.setRetryInterval(retryInterval); sf.setRetryIntervalMultiplier(retryMultiplier); sf.setReconnectAttempts(reconnectAttempts); + sf.setConfirmationWindowSize(1024 * 1024); ClientSession session = sf.createSession(false, true, true); @@ -245,12 +247,13 @@ sf.setRetryInterval(retryInterval); sf.setRetryIntervalMultiplier(retryMultiplier); sf.setReconnectAttempts(reconnectAttempts); + sf.setConfirmationWindowSize(1024 * 1024); ClientSession session = sf.createSession(false, true, true); ClientSession session2 = sf.createSession(false, true, true); - class MyFailureListener implements FailureListener + class MyFailureListener implements SessionFailureListener { volatile boolean failed; @@ -258,6 +261,10 @@ { failed = true; } + + public void beforeReconnect(HornetQException exception) + { + } } MyFailureListener listener = new MyFailureListener(); @@ -354,6 +361,7 @@ sf.setRetryInterval(retryInterval); sf.setRetryIntervalMultiplier(retryMultiplier); sf.setReconnectAttempts(reconnectAttempts); + sf.setConfirmationWindowSize(1024 * 1024); ClientSession session = sf.createSession(false, true, true); @@ -430,7 +438,8 @@ sf.setRetryInterval(retryInterval); sf.setRetryIntervalMultiplier(retryMultiplier); sf.setReconnectAttempts(reconnectAttempts); - + sf.setConfirmationWindowSize(1024 * 1024); + ClientSession session = sf.createSession(false, true, true); session.createQueue(ADDRESS, ADDRESS, null, false); @@ -497,6 +506,7 @@ sf.setRetryInterval(retryInterval); sf.setRetryIntervalMultiplier(retryMultiplier); sf.setReconnectAttempts(reconnectAttempts); + sf.setConfirmationWindowSize(1024 * 1024); ClientSession session = sf.createSession(false, true, true); @@ -588,6 +598,7 @@ sf.setRetryInterval(retryInterval); sf.setRetryIntervalMultiplier(retryMultiplier); sf.setReconnectAttempts(reconnectAttempts); + sf.setConfirmationWindowSize(1024 * 1024); ClientSession session = sf.createSession(false, true, true); @@ -666,6 +677,7 @@ sf.setRetryIntervalMultiplier(retryMultiplier); sf.setReconnectAttempts(reconnectAttempts); sf.setMaxRetryInterval(maxRetryInterval); + sf.setConfirmationWindowSize(1024 * 1024); ClientSession session = sf.createSession(false, true, true); Modified: trunk/tests/src/org/hornetq/tests/integration/jms/FloodServerTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/jms/FloodServerTest.java 2009-10-30 15:12:21 UTC (rev 8170) +++ trunk/tests/src/org/hornetq/tests/integration/jms/FloodServerTest.java 2009-10-30 17:41:19 UTC (rev 8171) @@ -21,7 +21,7 @@ import static org.hornetq.core.client.impl.ClientSessionFactoryImpl.DEFAULT_CONSUMER_WINDOW_SIZE; import static org.hornetq.core.client.impl.ClientSessionFactoryImpl.DEFAULT_MIN_LARGE_MESSAGE_SIZE; import static org.hornetq.core.client.impl.ClientSessionFactoryImpl.DEFAULT_PRODUCER_MAX_RATE; -import static org.hornetq.core.client.impl.ClientSessionFactoryImpl.DEFAULT_PRODUCER_WINDOW_SIZE; +import static org.hornetq.core.client.impl.ClientSessionFactoryImpl.DEFAULT_CONFIRMATION_WINDOW_SIZE; import static org.hornetq.core.client.impl.ClientSessionFactoryImpl.DEFAULT_SCHEDULED_THREAD_POOL_MAX_SIZE; import static org.hornetq.core.client.impl.ClientSessionFactoryImpl.DEFAULT_THREAD_POOL_MAX_SIZE; import static org.hornetq.core.client.impl.ClientSessionFactoryImpl.DEFAULT_USE_GLOBAL_POOLS; @@ -154,7 +154,7 @@ DEFAULT_MIN_LARGE_MESSAGE_SIZE, DEFAULT_CONSUMER_WINDOW_SIZE, DEFAULT_CONSUMER_MAX_RATE, - DEFAULT_PRODUCER_WINDOW_SIZE, + DEFAULT_CONFIRMATION_WINDOW_SIZE, DEFAULT_PRODUCER_MAX_RATE, false, false, Modified: trunk/tests/src/org/hornetq/tests/integration/jms/HornetQConnectionFactoryTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/jms/HornetQConnectionFactoryTest.java 2009-10-30 15:12:21 UTC (rev 8170) +++ trunk/tests/src/org/hornetq/tests/integration/jms/HornetQConnectionFactoryTest.java 2009-10-30 17:41:19 UTC (rev 8171) @@ -77,7 +77,7 @@ ClientSessionFactoryImpl.DEFAULT_MIN_LARGE_MESSAGE_SIZE, ClientSessionFactoryImpl.DEFAULT_CONSUMER_WINDOW_SIZE, ClientSessionFactoryImpl.DEFAULT_CONSUMER_MAX_RATE, - ClientSessionFactoryImpl.DEFAULT_PRODUCER_WINDOW_SIZE, + ClientSessionFactoryImpl.DEFAULT_CONFIRMATION_WINDOW_SIZE, ClientSessionFactoryImpl.DEFAULT_PRODUCER_MAX_RATE, ClientSessionFactoryImpl.DEFAULT_BLOCK_ON_ACKNOWLEDGE, ClientSessionFactoryImpl.DEFAULT_BLOCK_ON_PERSISTENT_SEND, @@ -147,7 +147,7 @@ ClientSessionFactoryImpl.DEFAULT_MIN_LARGE_MESSAGE_SIZE, ClientSessionFactoryImpl.DEFAULT_CONSUMER_WINDOW_SIZE, ClientSessionFactoryImpl.DEFAULT_CONSUMER_MAX_RATE, - ClientSessionFactoryImpl.DEFAULT_PRODUCER_WINDOW_SIZE, + ClientSessionFactoryImpl.DEFAULT_CONFIRMATION_WINDOW_SIZE, ClientSessionFactoryImpl.DEFAULT_PRODUCER_MAX_RATE, ClientSessionFactoryImpl.DEFAULT_BLOCK_ON_ACKNOWLEDGE, ClientSessionFactoryImpl.DEFAULT_BLOCK_ON_PERSISTENT_SEND, @@ -194,7 +194,7 @@ ClientSessionFactoryImpl.DEFAULT_MIN_LARGE_MESSAGE_SIZE, ClientSessionFactoryImpl.DEFAULT_CONSUMER_WINDOW_SIZE, ClientSessionFactoryImpl.DEFAULT_CONSUMER_MAX_RATE, - ClientSessionFactoryImpl.DEFAULT_PRODUCER_WINDOW_SIZE, + ClientSessionFactoryImpl.DEFAULT_CONFIRMATION_WINDOW_SIZE, ClientSessionFactoryImpl.DEFAULT_PRODUCER_MAX_RATE, ClientSessionFactoryImpl.DEFAULT_BLOCK_ON_ACKNOWLEDGE, ClientSessionFactoryImpl.DEFAULT_BLOCK_ON_PERSISTENT_SEND, @@ -242,7 +242,7 @@ ClientSessionFactoryImpl.DEFAULT_MIN_LARGE_MESSAGE_SIZE, ClientSessionFactoryImpl.DEFAULT_CONSUMER_WINDOW_SIZE, ClientSessionFactoryImpl.DEFAULT_CONSUMER_MAX_RATE, - ClientSessionFactoryImpl.DEFAULT_PRODUCER_WINDOW_SIZE, + ClientSessionFactoryImpl.DEFAULT_CONFIRMATION_WINDOW_SIZE, ClientSessionFactoryImpl.DEFAULT_PRODUCER_MAX_RATE, ClientSessionFactoryImpl.DEFAULT_BLOCK_ON_ACKNOWLEDGE, ClientSessionFactoryImpl.DEFAULT_BLOCK_ON_PERSISTENT_SEND, @@ -290,7 +290,7 @@ ClientSessionFactoryImpl.DEFAULT_MIN_LARGE_MESSAGE_SIZE, ClientSessionFactoryImpl.DEFAULT_CONSUMER_WINDOW_SIZE, ClientSessionFactoryImpl.DEFAULT_CONSUMER_MAX_RATE, - ClientSessionFactoryImpl.DEFAULT_PRODUCER_WINDOW_SIZE, + ClientSessionFactoryImpl.DEFAULT_CONFIRMATION_WINDOW_SIZE, ClientSessionFactoryImpl.DEFAULT_PRODUCER_MAX_RATE, ClientSessionFactoryImpl.DEFAULT_BLOCK_ON_ACKNOWLEDGE, ClientSessionFactoryImpl.DEFAULT_BLOCK_ON_PERSISTENT_SEND, @@ -335,7 +335,7 @@ int minLargeMessageSize = RandomUtil.randomPositiveInt(); int consumerWindowSize = RandomUtil.randomPositiveInt(); int consumerMaxRate = RandomUtil.randomPositiveInt(); - int producerWindowSize = RandomUtil.randomPositiveInt(); + int confirmationWindowSize = RandomUtil.randomPositiveInt(); int producerMaxRate = RandomUtil.randomPositiveInt(); boolean blockOnAcknowledge = RandomUtil.randomBoolean(); boolean blockOnPersistentSend = RandomUtil.randomBoolean(); @@ -363,7 +363,7 @@ cf.setMinLargeMessageSize(minLargeMessageSize); cf.setConsumerWindowSize(consumerWindowSize); cf.setConsumerMaxRate(consumerMaxRate); - cf.setProducerWindowSize(producerWindowSize); + cf.setConfirmationWindowSize(confirmationWindowSize); cf.setProducerMaxRate(producerMaxRate); cf.setBlockOnAcknowledge(blockOnAcknowledge); cf.setBlockOnPersistentSend(blockOnPersistentSend); @@ -391,7 +391,7 @@ assertEquals(minLargeMessageSize, cf.getMinLargeMessageSize()); assertEquals(consumerWindowSize, cf.getConsumerWindowSize()); assertEquals(consumerMaxRate, cf.getConsumerMaxRate()); - assertEquals(producerWindowSize, cf.getProducerWindowSize()); + assertEquals(confirmationWindowSize, cf.getConfirmationWindowSize()); assertEquals(producerMaxRate, cf.getProducerMaxRate()); assertEquals(blockOnAcknowledge, cf.isBlockOnAcknowledge()); assertEquals(blockOnPersistentSend, cf.isBlockOnPersistentSend()); @@ -428,7 +428,7 @@ int minLargeMessageSize = RandomUtil.randomPositiveInt(); int consumerWindowSize = RandomUtil.randomPositiveInt(); int consumerMaxRate = RandomUtil.randomPositiveInt(); - int producerWindowSize = RandomUtil.randomPositiveInt(); + int confirmationWindowSize = RandomUtil.randomPositiveInt(); int producerMaxRate = RandomUtil.randomPositiveInt(); boolean blockOnAcknowledge = RandomUtil.randomBoolean(); boolean blockOnPersistentSend = RandomUtil.randomBoolean(); @@ -548,7 +548,7 @@ } try { - cf.setProducerWindowSize(producerWindowSize); + cf.setConfirmationWindowSize(confirmationWindowSize); fail("Should throw exception"); } catch (IllegalStateException e) @@ -720,7 +720,7 @@ cf.getMinLargeMessageSize(); cf.getConsumerWindowSize(); cf.getConsumerMaxRate(); - cf.getProducerWindowSize(); + cf.getConfirmationWindowSize(); cf.getProducerMaxRate(); cf.isBlockOnAcknowledge(); cf.isBlockOnPersistentSend(); @@ -753,7 +753,7 @@ int minLargeMessageSize, int consumerWindowSize, int consumerMaxRate, - int producerWindowSize, + int confirmationWindowSize, int producerMaxRate, boolean blockOnAcknowledge, boolean blockOnPersistentSend, @@ -796,7 +796,7 @@ assertEquals(cf.getMinLargeMessageSize(), minLargeMessageSize); assertEquals(cf.getConsumerWindowSize(), consumerWindowSize); assertEquals(cf.getConsumerMaxRate(), consumerMaxRate); - assertEquals(cf.getProducerWindowSize(), producerWindowSize); + assertEquals(cf.getConfirmationWindowSize(), confirmationWindowSize); assertEquals(cf.getProducerMaxRate(), producerMaxRate); assertEquals(cf.isBlockOnAcknowledge(), blockOnAcknowledge); assertEquals(cf.isBlockOnPersistentSend(), blockOnPersistentSend); Modified: trunk/tests/src/org/hornetq/tests/integration/jms/ManualReconnectionToSingleServerTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/jms/ManualReconnectionToSingleServerTest.java 2009-10-30 15:12:21 UTC (rev 8170) +++ trunk/tests/src/org/hornetq/tests/integration/jms/ManualReconnectionToSingleServerTest.java 2009-10-30 17:41:19 UTC (rev 8171) @@ -27,7 +27,7 @@ import static org.hornetq.core.client.impl.ClientSessionFactoryImpl.DEFAULT_MIN_LARGE_MESSAGE_SIZE; import static org.hornetq.core.client.impl.ClientSessionFactoryImpl.DEFAULT_PRE_ACKNOWLEDGE; import static org.hornetq.core.client.impl.ClientSessionFactoryImpl.DEFAULT_PRODUCER_MAX_RATE; -import static org.hornetq.core.client.impl.ClientSessionFactoryImpl.DEFAULT_PRODUCER_WINDOW_SIZE; +import static org.hornetq.core.client.impl.ClientSessionFactoryImpl.DEFAULT_CONFIRMATION_WINDOW_SIZE; import static org.hornetq.core.client.impl.ClientSessionFactoryImpl.DEFAULT_SCHEDULED_THREAD_POOL_MAX_SIZE; import static org.hornetq.core.client.impl.ClientSessionFactoryImpl.DEFAULT_THREAD_POOL_MAX_SIZE; import static org.hornetq.core.client.impl.ClientSessionFactoryImpl.DEFAULT_USE_GLOBAL_POOLS; @@ -258,7 +258,7 @@ DEFAULT_MIN_LARGE_MESSAGE_SIZE, DEFAULT_CONSUMER_WINDOW_SIZE, DEFAULT_CONSUMER_MAX_RATE, - DEFAULT_PRODUCER_WINDOW_SIZE, + DEFAULT_CONFIRMATION_WINDOW_SIZE, DEFAULT_PRODUCER_MAX_RATE, DEFAULT_BLOCK_ON_ACKNOWLEDGE, DEFAULT_BLOCK_ON_PERSISTENT_SEND, Modified: trunk/tests/src/org/hornetq/tests/integration/jms/client/PreACKJMSTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/jms/client/PreACKJMSTest.java 2009-10-30 15:12:21 UTC (rev 8170) +++ trunk/tests/src/org/hornetq/tests/integration/jms/client/PreACKJMSTest.java 2009-10-30 17:41:19 UTC (rev 8171) @@ -27,7 +27,7 @@ import static org.hornetq.core.client.impl.ClientSessionFactoryImpl.DEFAULT_MAX_RETRY_INTERVAL; import static org.hornetq.core.client.impl.ClientSessionFactoryImpl.DEFAULT_MIN_LARGE_MESSAGE_SIZE; import static org.hornetq.core.client.impl.ClientSessionFactoryImpl.DEFAULT_PRODUCER_MAX_RATE; -import static org.hornetq.core.client.impl.ClientSessionFactoryImpl.DEFAULT_PRODUCER_WINDOW_SIZE; +import static org.hornetq.core.client.impl.ClientSessionFactoryImpl.DEFAULT_CONFIRMATION_WINDOW_SIZE; import static org.hornetq.core.client.impl.ClientSessionFactoryImpl.DEFAULT_SCHEDULED_THREAD_POOL_MAX_SIZE; import static org.hornetq.core.client.impl.ClientSessionFactoryImpl.DEFAULT_THREAD_POOL_MAX_SIZE; import static org.hornetq.core.client.impl.ClientSessionFactoryImpl.DEFAULT_USE_GLOBAL_POOLS; @@ -228,7 +228,7 @@ DEFAULT_MIN_LARGE_MESSAGE_SIZE, DEFAULT_CONSUMER_WINDOW_SIZE, DEFAULT_CONSUMER_MAX_RATE, - DEFAULT_PRODUCER_WINDOW_SIZE, + DEFAULT_CONFIRMATION_WINDOW_SIZE, DEFAULT_PRODUCER_MAX_RATE, DEFAULT_BLOCK_ON_ACKNOWLEDGE, DEFAULT_BLOCK_ON_PERSISTENT_SEND, Modified: trunk/tests/src/org/hornetq/tests/integration/jms/client/ResendTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/jms/client/ResendTest.java 2009-10-30 15:12:21 UTC (rev 8170) +++ trunk/tests/src/org/hornetq/tests/integration/jms/client/ResendTest.java 2009-10-30 17:41:19 UTC (rev 8171) @@ -27,7 +27,7 @@ import static org.hornetq.core.client.impl.ClientSessionFactoryImpl.DEFAULT_MIN_LARGE_MESSAGE_SIZE; import static org.hornetq.core.client.impl.ClientSessionFactoryImpl.DEFAULT_PRE_ACKNOWLEDGE; import static org.hornetq.core.client.impl.ClientSessionFactoryImpl.DEFAULT_PRODUCER_MAX_RATE; -import static org.hornetq.core.client.impl.ClientSessionFactoryImpl.DEFAULT_PRODUCER_WINDOW_SIZE; +import static org.hornetq.core.client.impl.ClientSessionFactoryImpl.DEFAULT_CONFIRMATION_WINDOW_SIZE; import static org.hornetq.core.client.impl.ClientSessionFactoryImpl.DEFAULT_SCHEDULED_THREAD_POOL_MAX_SIZE; import static org.hornetq.core.client.impl.ClientSessionFactoryImpl.DEFAULT_THREAD_POOL_MAX_SIZE; import static org.hornetq.core.client.impl.ClientSessionFactoryImpl.DEFAULT_USE_GLOBAL_POOLS; @@ -268,7 +268,7 @@ DEFAULT_MIN_LARGE_MESSAGE_SIZE, DEFAULT_CONSUMER_WINDOW_SIZE, DEFAULT_CONSUMER_MAX_RATE, - DEFAULT_PRODUCER_WINDOW_SIZE, + DEFAULT_CONFIRMATION_WINDOW_SIZE, DEFAULT_PRODUCER_MAX_RATE, DEFAULT_BLOCK_ON_ACKNOWLEDGE, DEFAULT_BLOCK_ON_PERSISTENT_SEND, Modified: trunk/tests/src/org/hornetq/tests/integration/jms/divert/DivertAndACKClientTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/jms/divert/DivertAndACKClientTest.java 2009-10-30 15:12:21 UTC (rev 8170) +++ trunk/tests/src/org/hornetq/tests/integration/jms/divert/DivertAndACKClientTest.java 2009-10-30 17:41:19 UTC (rev 8171) @@ -27,7 +27,7 @@ import static org.hornetq.core.client.impl.ClientSessionFactoryImpl.DEFAULT_MIN_LARGE_MESSAGE_SIZE; import static org.hornetq.core.client.impl.ClientSessionFactoryImpl.DEFAULT_PRE_ACKNOWLEDGE; import static org.hornetq.core.client.impl.ClientSessionFactoryImpl.DEFAULT_PRODUCER_MAX_RATE; -import static org.hornetq.core.client.impl.ClientSessionFactoryImpl.DEFAULT_PRODUCER_WINDOW_SIZE; +import static org.hornetq.core.client.impl.ClientSessionFactoryImpl.DEFAULT_CONFIRMATION_WINDOW_SIZE; import static org.hornetq.core.client.impl.ClientSessionFactoryImpl.DEFAULT_SCHEDULED_THREAD_POOL_MAX_SIZE; import static org.hornetq.core.client.impl.ClientSessionFactoryImpl.DEFAULT_THREAD_POOL_MAX_SIZE; import static org.hornetq.core.client.impl.ClientSessionFactoryImpl.DEFAULT_USE_GLOBAL_POOLS; @@ -167,7 +167,7 @@ DEFAULT_MIN_LARGE_MESSAGE_SIZE, DEFAULT_CONSUMER_WINDOW_SIZE, DEFAULT_CONSUMER_MAX_RATE, - DEFAULT_PRODUCER_WINDOW_SIZE, + DEFAULT_CONFIRMATION_WINDOW_SIZE, DEFAULT_PRODUCER_MAX_RATE, true, // this test needs to block on ACK DEFAULT_BLOCK_ON_PERSISTENT_SEND, Modified: trunk/tests/src/org/hornetq/tests/integration/jms/server/JMSServerDeployerTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/jms/server/JMSServerDeployerTest.java 2009-10-30 15:12:21 UTC (rev 8170) +++ trunk/tests/src/org/hornetq/tests/integration/jms/server/JMSServerDeployerTest.java 2009-10-30 17:41:19 UTC (rev 8171) @@ -194,7 +194,7 @@ assertEquals(5678, cf.getCallTimeout()); assertEquals(12345, cf.getConsumerWindowSize()); assertEquals(6789, cf.getConsumerMaxRate()); - assertEquals(123456, cf.getProducerWindowSize()); + assertEquals(123456, cf.getConfirmationWindowSize()); assertEquals(789, cf.getProducerMaxRate()); assertEquals(12, cf.getMinLargeMessageSize()); assertEquals("TestClientID", cf.getClientID()); Modified: trunk/tests/src/org/hornetq/tests/integration/jms/server/management/JMSServerControlTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/jms/server/management/JMSServerControlTest.java 2009-10-30 15:12:21 UTC (rev 8170) +++ trunk/tests/src/org/hornetq/tests/integration/jms/server/management/JMSServerControlTest.java 2009-10-30 17:41:19 UTC (rev 8171) @@ -433,7 +433,7 @@ ClientSessionFactoryImpl.DEFAULT_MIN_LARGE_MESSAGE_SIZE, ClientSessionFactoryImpl.DEFAULT_CONSUMER_WINDOW_SIZE, ClientSessionFactoryImpl.DEFAULT_CONSUMER_MAX_RATE, - ClientSessionFactoryImpl.DEFAULT_PRODUCER_WINDOW_SIZE, + ClientSessionFactoryImpl.DEFAULT_CONFIRMATION_WINDOW_SIZE, ClientSessionFactoryImpl.DEFAULT_PRODUCER_MAX_RATE, ClientSessionFactoryImpl.DEFAULT_BLOCK_ON_ACKNOWLEDGE, ClientSessionFactoryImpl.DEFAULT_BLOCK_ON_PERSISTENT_SEND, @@ -478,7 +478,7 @@ ClientSessionFactoryImpl.DEFAULT_MIN_LARGE_MESSAGE_SIZE, ClientSessionFactoryImpl.DEFAULT_CONSUMER_WINDOW_SIZE, ClientSessionFactoryImpl.DEFAULT_CONSUMER_MAX_RATE, - ClientSessionFactoryImpl.DEFAULT_PRODUCER_WINDOW_SIZE, + ClientSessionFactoryImpl.DEFAULT_CONFIRMATION_WINDOW_SIZE, ClientSessionFactoryImpl.DEFAULT_PRODUCER_MAX_RATE, ClientSessionFactoryImpl.DEFAULT_BLOCK_ON_ACKNOWLEDGE, ClientSessionFactoryImpl.DEFAULT_BLOCK_ON_PERSISTENT_SEND, Modified: trunk/tests/src/org/hornetq/tests/integration/jms/server/management/JMSServerControlUsingJMSTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/jms/server/management/JMSServerControlUsingJMSTest.java 2009-10-30 15:12:21 UTC (rev 8170) +++ trunk/tests/src/org/hornetq/tests/integration/jms/server/management/JMSServerControlUsingJMSTest.java 2009-10-30 17:41:19 UTC (rev 8171) @@ -104,7 +104,7 @@ int minLargeMessageSize, int consumerWindowSize, int consumerMaxRate, - int producerWindowSize, + int confirmationWindowSize, int producerMaxRate, boolean blockOnAcknowledge, boolean blockOnPersistentSend, @@ -137,7 +137,7 @@ minLargeMessageSize, consumerWindowSize, consumerMaxRate, - producerWindowSize, + confirmationWindowSize, producerMaxRate, blockOnAcknowledge, blockOnPersistentSend, @@ -171,7 +171,7 @@ int minLargeMessageSize, int consumerWindowSize, int consumerMaxRate, - int producerWindowSize, + int confirmationWindowSize, int producerMaxRate, boolean blockOnAcknowledge, boolean blockOnPersistentSend, @@ -204,7 +204,7 @@ minLargeMessageSize, consumerWindowSize, consumerMaxRate, - producerWindowSize, + confirmationWindowSize, producerMaxRate, blockOnAcknowledge, blockOnPersistentSend, @@ -387,7 +387,7 @@ int minLargeMessageSize, int consumerWindowSize, int consumerMaxRate, - int producerWindowSize, + int confirmationWindowSize, int producerMaxRate, boolean blockOnAcknowledge, boolean blockOnPersistentSend, @@ -421,7 +421,7 @@ minLargeMessageSize, consumerWindowSize, consumerMaxRate, - producerWindowSize, + confirmationWindowSize, producerMaxRate, blockOnAcknowledge, blockOnPersistentSend, @@ -456,7 +456,7 @@ int minLargeMessageSize, int consumerWindowSize, int consumerMaxRate, - int producerWindowSize, + int confirmationWindowSize, int producerMaxRate, boolean blockOnAcknowledge, boolean blockOnPersistentSend, @@ -490,7 +490,7 @@ minLargeMessageSize, consumerWindowSize, consumerMaxRate, - producerWindowSize, + confirmationWindowSize, producerMaxRate, blockOnAcknowledge, blockOnPersistentSend, Modified: trunk/tests/src/org/hornetq/tests/integration/largemessage/LargeMessageTestBase.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/largemessage/LargeMessageTestBase.java 2009-10-30 15:12:21 UTC (rev 8170) +++ trunk/tests/src/org/hornetq/tests/integration/largemessage/LargeMessageTestBase.java 2009-10-30 17:41:19 UTC (rev 8171) @@ -156,7 +156,7 @@ if (producerWindow > 0) { - sf.setProducerWindowSize(producerWindow); + sf.setConfirmationWindowSize(producerWindow); } sf.setMinLargeMessageSize(minSize); Modified: trunk/tests/src/org/hornetq/tests/integration/remoting/PingTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/remoting/PingTest.java 2009-10-30 15:12:21 UTC (rev 8170) +++ trunk/tests/src/org/hornetq/tests/integration/remoting/PingTest.java 2009-10-30 17:41:19 UTC (rev 8171) @@ -17,6 +17,7 @@ import org.hornetq.core.client.ClientSession; import org.hornetq.core.client.ClientSessionFactory; +import org.hornetq.core.client.SessionFailureListener; import org.hornetq.core.client.impl.ClientSessionFactoryImpl; import org.hornetq.core.client.impl.ClientSessionFactoryInternal; import org.hornetq.core.client.impl.FailoverManagerImpl; @@ -24,7 +25,6 @@ import org.hornetq.core.config.TransportConfiguration; import org.hornetq.core.exception.HornetQException; import org.hornetq.core.logging.Logger; -import org.hornetq.core.remoting.FailureListener; import org.hornetq.core.remoting.RemotingConnection; import org.hornetq.core.server.HornetQServer; import org.hornetq.tests.util.ServiceTestBase; @@ -71,7 +71,7 @@ super.tearDown(); } - class Listener implements FailureListener + class Listener implements SessionFailureListener { volatile HornetQException me; @@ -84,6 +84,10 @@ { return me; } + + public void beforeReconnect(HornetQException exception) + { + } }; /* @@ -318,10 +322,8 @@ serverConn.addFailureListener(serverListener); - //((RemotingServiceImpl)server.getRemotingService()).stopPingingForConnectionID(serverConn.getID()); - //Setting the handler to null will prevent server sending pings back to client - serverConn.getChannel(0, -1, false).setHandler(null); + serverConn.getChannel(0, -1).setHandler(null); for (int i = 0; i < 1000; i++) { Modified: trunk/tests/src/org/hornetq/tests/integration/replication/ReplicationTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/replication/ReplicationTest.java 2009-10-30 15:12:21 UTC (rev 8170) +++ trunk/tests/src/org/hornetq/tests/integration/replication/ReplicationTest.java 2009-10-30 17:41:19 UTC (rev 8171) @@ -242,7 +242,7 @@ serverMsg.setDestination(new SimpleString("tttt")); HornetQBuffer buffer = ChannelBuffers.dynamicBuffer(100); - serverMsg.encodeProperties(buffer); + serverMsg.encodeHeadersAndProperties(buffer); manager.largeMessageBegin(500); 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 2009-10-30 15:12:21 UTC (rev 8170) +++ trunk/tests/src/org/hornetq/tests/unit/core/postoffice/impl/BindingsImplTest.java 2009-10-30 17:41:19 UTC (rev 8171) @@ -22,6 +22,7 @@ import org.hornetq.core.exception.HornetQException; import org.hornetq.core.filter.Filter; +import org.hornetq.core.paging.PagingStore; import org.hornetq.core.postoffice.Binding; import org.hornetq.core.postoffice.BindingType; import org.hornetq.core.postoffice.impl.BindingsImpl; @@ -442,7 +443,7 @@ /* (non-Javadoc) * @see org.hornetq.core.message.Message#decodeProperties(org.hornetq.core.remoting.spi.HornetQBuffer) */ - public void decodeProperties(final HornetQBuffer buffer) + public void decodeHeadersAndProperties(final HornetQBuffer buffer) { } @@ -474,7 +475,7 @@ /* (non-Javadoc) * @see org.hornetq.core.message.Message#encodeProperties(org.hornetq.core.remoting.spi.HornetQBuffer) */ - public void encodeProperties(final HornetQBuffer buffer) + public void encodeHeadersAndProperties(final HornetQBuffer buffer) { } @@ -554,7 +555,7 @@ /* (non-Javadoc) * @see org.hornetq.core.message.Message#getPropertiesEncodeSize() */ - public int getPropertiesEncodeSize() + public int getHeadersAndPropertiesEncodeSize() { return 0; @@ -882,6 +883,24 @@ return null; } + /* (non-Javadoc) + * @see org.hornetq.core.server.ServerMessage#decrementRefCount(org.hornetq.core.paging.PagingStore, org.hornetq.core.server.MessageReference) + */ + public int decrementRefCount(PagingStore pagingStore, MessageReference reference) throws Exception + { + // TODO Auto-generated method stub + return 0; + } + + /* (non-Javadoc) + * @see org.hornetq.core.server.ServerMessage#incrementRefCount(org.hornetq.core.paging.PagingStore, org.hornetq.core.server.MessageReference) + */ + public int incrementRefCount(PagingStore pagingStore, MessageReference reference) throws Exception + { + // TODO Auto-generated method stub + return 0; + } + } class FakeFilter implements Filter Modified: trunk/tests/src/org/hornetq/tests/unit/core/server/impl/fakes/FakePostOffice.java =================================================================== --- trunk/tests/src/org/hornetq/tests/unit/core/server/impl/fakes/FakePostOffice.java 2009-10-30 15:12:21 UTC (rev 8170) +++ trunk/tests/src/org/hornetq/tests/unit/core/server/impl/fakes/FakePostOffice.java 2009-10-30 17:41:19 UTC (rev 8171) @@ -137,39 +137,41 @@ return null; } + + /* (non-Javadoc) - * @see org.hornetq.core.postoffice.PostOffice#reroute(org.hornetq.core.server.ServerMessage, org.hornetq.core.server.Queue, org.hornetq.core.transaction.Transaction) + * @see org.hornetq.core.postoffice.PostOffice#sendQueueInfoToQueue(org.hornetq.utils.SimpleString, org.hornetq.utils.SimpleString) */ - public MessageReference reroute(ServerMessage message, Queue queue, Transaction tx) throws Exception + public void sendQueueInfoToQueue(SimpleString queueName, SimpleString address) throws Exception { // TODO Auto-generated method stub - return null; + } /* (non-Javadoc) - * @see org.hornetq.core.postoffice.PostOffice#route(org.hornetq.core.server.ServerMessage, org.hornetq.core.server.RoutingContext) + * @see org.hornetq.core.postoffice.PostOffice#route(org.hornetq.core.server.ServerMessage) */ - public void route(ServerMessage message, RoutingContext context) throws Exception + public void route(ServerMessage message) throws Exception { // TODO Auto-generated method stub } - /* (non-Javadoc) - * @see org.hornetq.core.postoffice.PostOffice#sendQueueInfoToQueue(org.hornetq.utils.SimpleString, org.hornetq.utils.SimpleString) - */ - public void sendQueueInfoToQueue(SimpleString queueName, SimpleString address) throws Exception + public boolean redistribute(ServerMessage message, Queue originatingQueue, Transaction tx) throws Exception { // TODO Auto-generated method stub - + return false; } - /* (non-Javadoc) - * @see org.hornetq.core.postoffice.PostOffice#route(org.hornetq.core.server.ServerMessage) - */ - public void route(ServerMessage message) throws Exception + public MessageReference reroute(ServerMessage message, Queue queue, Transaction tx) throws Exception { // TODO Auto-generated method stub + return null; + } + + public void route(ServerMessage message, Transaction tx) throws Exception + { + // TODO Auto-generated method stub } Modified: trunk/tests/src/org/hornetq/tests/unit/core/settings/impl/AddressSettingsTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/unit/core/settings/impl/AddressSettingsTest.java 2009-10-30 15:12:21 UTC (rev 8170) +++ trunk/tests/src/org/hornetq/tests/unit/core/settings/impl/AddressSettingsTest.java 2009-10-30 17:41:19 UTC (rev 8171) @@ -13,6 +13,7 @@ package org.hornetq.tests.unit.core.settings.impl; +import org.hornetq.core.settings.impl.AddressFullMessagePolicy; import org.hornetq.core.settings.impl.AddressSettings; import org.hornetq.tests.util.UnitTestCase; import org.hornetq.utils.SimpleString; @@ -46,7 +47,7 @@ addressSettingsToMerge.setDeadLetterAddress(DLQ); addressSettingsToMerge.setExpiryAddress(exp); addressSettingsToMerge.setMaxDeliveryAttempts(1000); - addressSettingsToMerge.setDropMessagesWhenFull(true); + addressSettingsToMerge.setAddressFullMessagePolicy(AddressFullMessagePolicy.DROP); addressSettingsToMerge.setMaxSizeBytes(1001); addressSettingsToMerge.setMessageCounterHistoryDayLimit(1002); addressSettingsToMerge.setRedeliveryDelay((long)1003); @@ -61,7 +62,7 @@ assertEquals(addressSettings.getMessageCounterHistoryDayLimit(), 1002); assertEquals(addressSettings.getRedeliveryDelay(), 1003); assertEquals(addressSettings.getPageSizeBytes(), 1004); - assertTrue(addressSettings.isDropMessagesWhenFull()); + assertEquals(AddressFullMessagePolicy.DROP, addressSettings.getAddressFullMessagePolicy()); } public void testMultipleMerge() @@ -75,6 +76,7 @@ addressSettingsToMerge.setMaxDeliveryAttempts(1000); addressSettingsToMerge.setMaxSizeBytes(1001); addressSettingsToMerge.setMessageCounterHistoryDayLimit(1002); + addressSettingsToMerge.setAddressFullMessagePolicy(AddressFullMessagePolicy.DROP); addressSettings.merge(addressSettingsToMerge); AddressSettings addressSettingsToMerge2 = new AddressSettings(); @@ -92,6 +94,7 @@ assertEquals(addressSettings.getMaxSizeBytes(), 1001); assertEquals(addressSettings.getMessageCounterHistoryDayLimit(), 1002); assertEquals(addressSettings.getRedeliveryDelay(), 2003); + assertEquals(AddressFullMessagePolicy.DROP, addressSettings.getAddressFullMessagePolicy()); } public void testMultipleMergeAll() @@ -104,6 +107,7 @@ addressSettingsToMerge.setExpiryAddress(exp); addressSettingsToMerge.setMaxSizeBytes(1001); addressSettingsToMerge.setRedeliveryDelay((long)1003); + addressSettingsToMerge.setAddressFullMessagePolicy(AddressFullMessagePolicy.DROP); addressSettings.merge(addressSettingsToMerge); AddressSettings addressSettingsToMerge2 = new AddressSettings(); @@ -115,6 +119,7 @@ addressSettingsToMerge2.setMaxSizeBytes(2001); addressSettingsToMerge2.setMessageCounterHistoryDayLimit(2002); addressSettingsToMerge2.setRedeliveryDelay((long)2003); + addressSettingsToMerge.setAddressFullMessagePolicy(AddressFullMessagePolicy.PAGE); addressSettings.merge(addressSettingsToMerge2); assertEquals(addressSettings.getDistributionPolicy().getClass(), AddressSettings.DEFAULT_DISTRIBUTION_POLICY_CLASS); @@ -125,5 +130,6 @@ assertEquals(addressSettings.getMaxSizeBytes(), 1001); assertEquals(addressSettings.getMessageCounterHistoryDayLimit(), 2002); assertEquals(addressSettings.getRedeliveryDelay(),1003); + assertEquals(AddressFullMessagePolicy.DROP, addressSettings.getAddressFullMessagePolicy()); } } Modified: trunk/tests/src/org/hornetq/tests/unit/ra/ResourceAdapterTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/unit/ra/ResourceAdapterTest.java 2009-10-30 15:12:21 UTC (rev 8170) +++ trunk/tests/src/org/hornetq/tests/unit/ra/ResourceAdapterTest.java 2009-10-30 17:41:19 UTC (rev 8171) @@ -83,7 +83,7 @@ assertEquals(factory.getDupsOKBatchSize(), ClientSessionFactoryImpl.DEFAULT_ACK_BATCH_SIZE); assertEquals(factory.getMinLargeMessageSize(), ClientSessionFactoryImpl.DEFAULT_MIN_LARGE_MESSAGE_SIZE); assertEquals(factory.getProducerMaxRate(), ClientSessionFactoryImpl.DEFAULT_PRODUCER_MAX_RATE); - assertEquals(factory.getProducerWindowSize(), ClientSessionFactoryImpl.DEFAULT_PRODUCER_WINDOW_SIZE); + assertEquals(factory.getConfirmationWindowSize(), ClientSessionFactoryImpl.DEFAULT_CONFIRMATION_WINDOW_SIZE); assertEquals(factory.getReconnectAttempts(), ClientSessionFactoryImpl.DEFAULT_RECONNECT_ATTEMPTS); assertEquals(factory.getRetryInterval(), ClientSessionFactoryImpl.DEFAULT_RETRY_INTERVAL); assertEquals(factory.getRetryIntervalMultiplier(), ClientSessionFactoryImpl.DEFAULT_RETRY_INTERVAL_MULTIPLIER); @@ -127,7 +127,7 @@ assertEquals(factory.getDupsOKBatchSize(), ClientSessionFactoryImpl.DEFAULT_ACK_BATCH_SIZE); assertEquals(factory.getMinLargeMessageSize(), ClientSessionFactoryImpl.DEFAULT_MIN_LARGE_MESSAGE_SIZE); assertEquals(factory.getProducerMaxRate(), ClientSessionFactoryImpl.DEFAULT_PRODUCER_MAX_RATE); - assertEquals(factory.getProducerWindowSize(), ClientSessionFactoryImpl.DEFAULT_PRODUCER_WINDOW_SIZE); + assertEquals(factory.getConfirmationWindowSize(), ClientSessionFactoryImpl.DEFAULT_CONFIRMATION_WINDOW_SIZE); assertEquals(factory.getReconnectAttempts(), ClientSessionFactoryImpl.DEFAULT_RECONNECT_ATTEMPTS); assertEquals(factory.getRetryInterval(), ClientSessionFactoryImpl.DEFAULT_RETRY_INTERVAL); assertEquals(factory.getRetryIntervalMultiplier(), ClientSessionFactoryImpl.DEFAULT_RETRY_INTERVAL_MULTIPLIER); @@ -165,7 +165,7 @@ ra.setMinLargeMessageSize(10); ra.setPreAcknowledge(!ClientSessionFactoryImpl.DEFAULT_PRE_ACKNOWLEDGE); ra.setProducerMaxRate(11); - ra.setProducerWindowSize(12); + ra.setConfirmationWindowSize(12); ra.setReconnectAttempts(13); ra.setRetryInterval(14l); ra.setRetryIntervalMultiplier(15d); @@ -188,7 +188,7 @@ assertEquals(factory.getDupsOKBatchSize(), 8); assertEquals(factory.getMinLargeMessageSize(), 10); assertEquals(factory.getProducerMaxRate(), 11); - assertEquals(factory.getProducerWindowSize(), 12); + assertEquals(factory.getConfirmationWindowSize(), 12); assertEquals(factory.getReconnectAttempts(), 13); assertEquals(factory.getRetryInterval(), 14); assertEquals(factory.getRetryIntervalMultiplier(), 15d); @@ -227,7 +227,7 @@ connectionFactoryProperties.setMinLargeMessageSize(10); connectionFactoryProperties.setPreAcknowledge(!ClientSessionFactoryImpl.DEFAULT_PRE_ACKNOWLEDGE); connectionFactoryProperties.setProducerMaxRate(11); - connectionFactoryProperties.setProducerWindowSize(12); + connectionFactoryProperties.setConfirmationWindowSize(12); connectionFactoryProperties.setReconnectAttempts(13); connectionFactoryProperties.setRetryInterval(14l); connectionFactoryProperties.setRetryIntervalMultiplier(15d); @@ -250,7 +250,7 @@ assertEquals(factory.getDupsOKBatchSize(), 8); assertEquals(factory.getMinLargeMessageSize(), 10); assertEquals(factory.getProducerMaxRate(), 11); - assertEquals(factory.getProducerWindowSize(), 12); + assertEquals(factory.getConfirmationWindowSize(), 12); assertEquals(factory.getReconnectAttempts(), 13); assertEquals(factory.getRetryInterval(), 14); assertEquals(factory.getRetryIntervalMultiplier(), 15d); Modified: trunk/tests/src/org/hornetq/tests/util/JMSTestBase.java =================================================================== --- trunk/tests/src/org/hornetq/tests/util/JMSTestBase.java 2009-10-30 15:12:21 UTC (rev 8170) +++ trunk/tests/src/org/hornetq/tests/util/JMSTestBase.java 2009-10-30 17:41:19 UTC (rev 8171) @@ -28,7 +28,7 @@ import static org.hornetq.core.client.impl.ClientSessionFactoryImpl.DEFAULT_MIN_LARGE_MESSAGE_SIZE; import static org.hornetq.core.client.impl.ClientSessionFactoryImpl.DEFAULT_PRE_ACKNOWLEDGE; import static org.hornetq.core.client.impl.ClientSessionFactoryImpl.DEFAULT_PRODUCER_MAX_RATE; -import static org.hornetq.core.client.impl.ClientSessionFactoryImpl.DEFAULT_PRODUCER_WINDOW_SIZE; +import static org.hornetq.core.client.impl.ClientSessionFactoryImpl.DEFAULT_CONFIRMATION_WINDOW_SIZE; import static org.hornetq.core.client.impl.ClientSessionFactoryImpl.DEFAULT_SCHEDULED_THREAD_POOL_MAX_SIZE; import static org.hornetq.core.client.impl.ClientSessionFactoryImpl.DEFAULT_THREAD_POOL_MAX_SIZE; import static org.hornetq.core.client.impl.ClientSessionFactoryImpl.DEFAULT_USE_GLOBAL_POOLS; @@ -207,7 +207,7 @@ DEFAULT_MIN_LARGE_MESSAGE_SIZE, DEFAULT_CONSUMER_WINDOW_SIZE, DEFAULT_CONSUMER_MAX_RATE, - DEFAULT_PRODUCER_WINDOW_SIZE, + DEFAULT_CONFIRMATION_WINDOW_SIZE, DEFAULT_PRODUCER_MAX_RATE, DEFAULT_BLOCK_ON_ACKNOWLEDGE, DEFAULT_BLOCK_ON_PERSISTENT_SEND,