Author: clebert.suconic(a)jboss.com Date: 2011-11-07 19:45:05 -0500 (Mon, 07 Nov 2011) New Revision: 11666 Modified: branches/Branch_2_2_AS7/src/config/common/schema/hornetq-configuration.xsd branches/Branch_2_2_AS7/src/main/org/hornetq/api/core/client/ServerLocator.java branches/Branch_2_2_AS7/src/main/org/hornetq/core/client/impl/ClientConsumerImpl.java branches/Branch_2_2_AS7/src/main/org/hornetq/core/client/impl/ClientProducerCreditManagerImpl.java branches/Branch_2_2_AS7/src/main/org/hornetq/core/client/impl/ClientProducerCredits.java branches/Branch_2_2_AS7/src/main/org/hornetq/core/client/impl/ClientProducerCreditsImpl.java branches/Branch_2_2_AS7/src/main/org/hornetq/core/client/impl/ClientSessionFactoryImpl.java branches/Branch_2_2_AS7/src/main/org/hornetq/core/client/impl/ClientSessionImpl.java branches/Branch_2_2_AS7/src/main/org/hornetq/core/client/impl/ServerLocatorImpl.java branches/Branch_2_2_AS7/src/main/org/hornetq/core/client/impl/Topology.java branches/Branch_2_2_AS7/src/main/org/hornetq/core/config/BridgeConfiguration.java branches/Branch_2_2_AS7/src/main/org/hornetq/core/config/ClusterConnectionConfiguration.java branches/Branch_2_2_AS7/src/main/org/hornetq/core/deployers/impl/FileConfigurationParser.java branches/Branch_2_2_AS7/src/main/org/hornetq/core/journal/SequentialFile.java branches/Branch_2_2_AS7/src/main/org/hornetq/core/journal/impl/AIOSequentialFile.java branches/Branch_2_2_AS7/src/main/org/hornetq/core/journal/impl/AbstractSequentialFile.java branches/Branch_2_2_AS7/src/main/org/hornetq/core/journal/impl/NIOSequentialFile.java branches/Branch_2_2_AS7/src/main/org/hornetq/core/management/impl/HornetQServerControlImpl.java branches/Branch_2_2_AS7/src/main/org/hornetq/core/paging/impl/PageImpl.java branches/Branch_2_2_AS7/src/main/org/hornetq/core/paging/impl/PagingManagerImpl.java branches/Branch_2_2_AS7/src/main/org/hornetq/core/paging/impl/PagingStoreImpl.java branches/Branch_2_2_AS7/src/main/org/hornetq/core/persistence/impl/journal/JournalStorageManager.java branches/Branch_2_2_AS7/src/main/org/hornetq/core/persistence/impl/journal/LargeServerMessageImpl.java branches/Branch_2_2_AS7/src/main/org/hornetq/core/persistence/impl/journal/PrintData.java branches/Branch_2_2_AS7/src/main/org/hornetq/core/postoffice/impl/BindingsImpl.java branches/Branch_2_2_AS7/src/main/org/hornetq/core/postoffice/impl/PostOfficeImpl.java branches/Branch_2_2_AS7/src/main/org/hornetq/core/postoffice/impl/SimpleAddressManager.java branches/Branch_2_2_AS7/src/main/org/hornetq/core/protocol/core/ServerSessionPacketHandler.java branches/Branch_2_2_AS7/src/main/org/hornetq/core/protocol/core/impl/CoreProtocolManager.java branches/Branch_2_2_AS7/src/main/org/hornetq/core/remoting/impl/netty/NettyConnection.java branches/Branch_2_2_AS7/src/main/org/hornetq/core/server/cluster/Bridge.java branches/Branch_2_2_AS7/src/main/org/hornetq/core/server/cluster/ClusterConnection.java branches/Branch_2_2_AS7/src/main/org/hornetq/core/server/cluster/impl/BridgeImpl.java branches/Branch_2_2_AS7/src/main/org/hornetq/core/server/cluster/impl/ClusterConnectionBridge.java branches/Branch_2_2_AS7/src/main/org/hornetq/core/server/cluster/impl/ClusterConnectionImpl.java branches/Branch_2_2_AS7/src/main/org/hornetq/core/server/cluster/impl/ClusterManagerImpl.java branches/Branch_2_2_AS7/src/main/org/hornetq/core/server/cluster/impl/RemoteQueueBindingImpl.java branches/Branch_2_2_AS7/src/main/org/hornetq/core/server/impl/DivertImpl.java branches/Branch_2_2_AS7/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java branches/Branch_2_2_AS7/src/main/org/hornetq/core/server/impl/QueueImpl.java branches/Branch_2_2_AS7/src/main/org/hornetq/jms/client/HornetQConnectionFactory.java branches/Branch_2_2_AS7/src/main/org/hornetq/jms/server/recovery/HornetQXAResourceWrapper.java branches/Branch_2_2_AS7/src/main/org/hornetq/jms/server/recovery/XARecoveryConfig.java branches/Branch_2_2_AS7/src/main/org/hornetq/ra/HornetQRAManagedConnectionFactory.java branches/Branch_2_2_AS7/src/main/org/hornetq/ra/HornetQResourceAdapter.java branches/Branch_2_2_AS7/src/main/org/hornetq/ra/recovery/RecoveryManager.java branches/Branch_2_2_AS7/tests/config/ConfigurationTest-full-config.xml branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/client/IncompatibleVersionTest.java branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/client/InterruptedLargeMessageTest.java branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/client/JournalCrashTest.java branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/client/LargeMessageCompressTest.java branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/client/LargeMessageTest.java branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/client/PagingOrderTest.java branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/client/PagingTest.java branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/cluster/bridge/BridgeReconnectTest.java branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/cluster/bridge/BridgeStartTest.java branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/cluster/bridge/BridgeTest.java branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/cluster/bridge/BridgeWithDiscoveryGroupStartTest.java branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/cluster/bridge/BridgeWithPagingTest.java branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/cluster/distribution/ClusterTestBase.java branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/cluster/distribution/ClusteredRequestResponseTest.java branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/cluster/distribution/MessageRedistributionTest.java branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/cluster/distribution/TwoWayTwoNodeClusterTest.java branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/cluster/topology/HAClientTopologyWithDiscoveryTest.java branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/cluster/topology/IsolatedTopologyTest.java branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/cluster/topology/TopologyClusterTestBase.java branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/largemessage/LargeMessageTestBase.java branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/management/BridgeControlTest.java branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/management/BridgeControlUsingCoreTest.java branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/ra/HornetQMessageHandlerSecurityTest.java branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/ra/HornetQMessageHandlerTest.java branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/ra/HornetQMessageHandlerXATest.java branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/ra/ResourceAdapterTest.java branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/unit/core/config/impl/FileConfigurationTest.java branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/unit/core/journal/impl/fakes/FakeSequentialFileFactory.java branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/unit/core/persistence/impl/BatchIDGeneratorUnitTest.java branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/util/UnitTestCase.java Log: merging changes from EAP Branch Modified: branches/Branch_2_2_AS7/src/config/common/schema/hornetq-configuration.xsd =================================================================== --- branches/Branch_2_2_AS7/src/config/common/schema/hornetq-configuration.xsd 2011-11-07 04:32:51 UTC (rev 11665) +++ branches/Branch_2_2_AS7/src/config/common/schema/hornetq-configuration.xsd 2011-11-08 00:45:05 UTC (rev 11666) @@ -313,6 +313,8 @@ </xsd:element> <xsd:element maxOccurs="1" minOccurs="0" name="transformer-class-name" type="xsd:string"> </xsd:element> + <xsd:element maxOccurs="1" minOccurs="0" name="min-large-message-size" type="xsd:int"> + </xsd:element> <xsd:element maxOccurs="1" minOccurs="0" name="check-period" type="xsd:long"> </xsd:element> <xsd:element maxOccurs="1" minOccurs="0" name="connection-ttl" type="xsd:long"> @@ -364,6 +366,8 @@ </xsd:element> <xsd:element maxOccurs="1" minOccurs="0" name="connection-ttl" type="xsd:long"> </xsd:element> + <xsd:element maxOccurs="1" minOccurs="0" name="min-large-message-size" type="xsd:int"> + </xsd:element> <xsd:element maxOccurs="1" minOccurs="0" name="call-timeout" type="xsd:long"> </xsd:element> <xsd:element maxOccurs="1" minOccurs="0" name="retry-interval" type="xsd:long"> Modified: branches/Branch_2_2_AS7/src/main/org/hornetq/api/core/client/ServerLocator.java =================================================================== --- branches/Branch_2_2_AS7/src/main/org/hornetq/api/core/client/ServerLocator.java 2011-11-07 04:32:51 UTC (rev 11665) +++ branches/Branch_2_2_AS7/src/main/org/hornetq/api/core/client/ServerLocator.java 2011-11-08 00:45:05 UTC (rev 11666) @@ -54,6 +54,15 @@ * Create a ClientSessionFactory to a specific server. The server must already be known about by this ServerLocator. * This method allows the user to make a connection to a specific server bypassing any load balancing policy in force * @param transportConfiguration + * @return The ClientSesionFactory or null if the node is not present on the topology + * @throws Exception if a failure happened in creating the ClientSessionFactory or the ServerLocator does not know about the passed in transportConfiguration + */ + ClientSessionFactory createSessionFactory(final String nodeID) throws Exception; + + /** + * Create a ClientSessionFactory to a specific server. The server must already be known about by this ServerLocator. + * This method allows the user to make a connection to a specific server bypassing any load balancing policy in force + * @param transportConfiguration * @return The ClientSesionFactory * @throws Exception if a failure happened in creating the ClientSessionFactory or the ServerLocator does not know about the passed in transportConfiguration */ Modified: branches/Branch_2_2_AS7/src/main/org/hornetq/core/client/impl/ClientConsumerImpl.java =================================================================== --- branches/Branch_2_2_AS7/src/main/org/hornetq/core/client/impl/ClientConsumerImpl.java 2011-11-07 04:32:51 UTC (rev 11665) +++ branches/Branch_2_2_AS7/src/main/org/hornetq/core/client/impl/ClientConsumerImpl.java 2011-11-08 00:45:05 UTC (rev 11666) @@ -14,6 +14,8 @@ package org.hornetq.core.client.impl; import java.io.File; +import java.security.AccessController; +import java.security.PrivilegedAction; import java.util.Iterator; import java.util.concurrent.Executor; @@ -116,6 +118,8 @@ private final SessionQueueQueryResponseMessage queueInfo; private volatile boolean ackIndividually; + + private final ClassLoader contextClassLoader; // Constructors // --------------------------------------------------------------------------------- @@ -130,7 +134,8 @@ final TokenBucketLimiter rateLimiter, final Executor executor, final Channel channel, - final SessionQueueQueryResponseMessage queueInfo) + final SessionQueueQueryResponseMessage queueInfo, + final ClassLoader contextClassLoader) { this.id = id; @@ -153,6 +158,8 @@ this.ackBatchSize = ackBatchSize; this.queueInfo = queueInfo; + + this.contextClassLoader = contextClassLoader; } // ClientConsumer implementation @@ -861,7 +868,7 @@ { return; } - + session.workDone(); // We pull the message from the buffer from inside the Runnable so we can ensure priority @@ -894,6 +901,8 @@ return; } + + boolean expired = message.isExpired(); flowControlBeforeConsumption(message); @@ -906,8 +915,33 @@ { ClientConsumerImpl.log.trace("Calling handler.onMessage"); } - theHandler.onMessage(message); + final ClassLoader originalLoader = AccessController.doPrivileged(new PrivilegedAction<ClassLoader>() + { + public ClassLoader run() + { + ClassLoader originalLoader = Thread.currentThread().getContextClassLoader(); + Thread.currentThread().setContextClassLoader(contextClassLoader); + + return originalLoader; + } + }); + try + { + theHandler.onMessage(message); + } + finally + { + AccessController.doPrivileged(new PrivilegedAction<Object>() + { + public Object run() + { + Thread.currentThread().setContextClassLoader(originalLoader); + return null; + } + }); + } + if (ClientConsumerImpl.trace) { ClientConsumerImpl.log.trace("Handler.onMessage done"); Modified: branches/Branch_2_2_AS7/src/main/org/hornetq/core/client/impl/ClientProducerCreditManagerImpl.java =================================================================== --- branches/Branch_2_2_AS7/src/main/org/hornetq/core/client/impl/ClientProducerCreditManagerImpl.java 2011-11-07 04:32:51 UTC (rev 11665) +++ branches/Branch_2_2_AS7/src/main/org/hornetq/core/client/impl/ClientProducerCreditManagerImpl.java 2011-11-08 00:45:05 UTC (rev 11666) @@ -48,29 +48,44 @@ this.windowSize = windowSize; } - public synchronized ClientProducerCredits getCredits(final SimpleString address, final boolean anon) + public ClientProducerCredits getCredits(final SimpleString address, final boolean anon) { - ClientProducerCredits credits = producerCredits.get(address); - - if (credits == null) + boolean needInit = false; + ClientProducerCredits credits; + + synchronized(this) { - // Doesn't need to be fair since session is single threaded - credits = new ClientProducerCreditsImpl(session, address, windowSize); - - producerCredits.put(address, credits); + credits = producerCredits.get(address); + + if (credits == null) + { + // Doesn't need to be fair since session is single threaded + credits = new ClientProducerCreditsImpl(session, address, windowSize); + needInit = true; + + producerCredits.put(address, credits); + } + + if (!anon) + { + credits.incrementRefCount(); + + // Remove from anon credits (if there) + unReferencedCredits.remove(address); + } + else + { + addToUnReferencedCache(address, credits); + } } - - if (!anon) + + // The init is done outside of the lock + // otherwise packages may arrive with flow control + // while this is still sending requests causing a dead lock + if (needInit) { - credits.incrementRefCount(); - - // Remove from anon credits (if there) - unReferencedCredits.remove(address); + credits.init(); } - else - { - addToUnReferencedCache(address, credits); - } return credits; } Modified: branches/Branch_2_2_AS7/src/main/org/hornetq/core/client/impl/ClientProducerCredits.java =================================================================== --- branches/Branch_2_2_AS7/src/main/org/hornetq/core/client/impl/ClientProducerCredits.java 2011-11-07 04:32:51 UTC (rev 11665) +++ branches/Branch_2_2_AS7/src/main/org/hornetq/core/client/impl/ClientProducerCredits.java 2011-11-08 00:45:05 UTC (rev 11666) @@ -27,6 +27,8 @@ void receiveCredits(int credits); boolean isBlocked(); + + void init(); void reset(); Modified: branches/Branch_2_2_AS7/src/main/org/hornetq/core/client/impl/ClientProducerCreditsImpl.java =================================================================== --- branches/Branch_2_2_AS7/src/main/org/hornetq/core/client/impl/ClientProducerCreditsImpl.java 2011-11-07 04:32:51 UTC (rev 11665) +++ branches/Branch_2_2_AS7/src/main/org/hornetq/core/client/impl/ClientProducerCreditsImpl.java 2011-11-08 00:45:05 UTC (rev 11666) @@ -58,7 +58,10 @@ // Doesn't need to be fair since session is single threaded semaphore = new Semaphore(0, false); - + } + + public void init() + { // 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); Modified: branches/Branch_2_2_AS7/src/main/org/hornetq/core/client/impl/ClientSessionFactoryImpl.java =================================================================== --- branches/Branch_2_2_AS7/src/main/org/hornetq/core/client/impl/ClientSessionFactoryImpl.java 2011-11-07 04:32:51 UTC (rev 11665) +++ branches/Branch_2_2_AS7/src/main/org/hornetq/core/client/impl/ClientSessionFactoryImpl.java 2011-11-08 00:45:05 UTC (rev 11666) @@ -885,9 +885,8 @@ } // Should never get here - throw new IllegalStateException("Internal Error! ClientSessionFactoryImpl::createSessionInternal " + - "just reached a condition that was not supposed to happen. " + - "Please inform this condition to the HornetQ team"); + throw new IllegalStateException("Internal Error! ClientSessionFactoryImpl::createSessionInternal " + "just reached a condition that was not supposed to happen. " + + "Please inform this condition to the HornetQ team"); } private void callFailureListeners(final HornetQException me, final boolean afterReconnect, final boolean failedOver) @@ -1293,7 +1292,9 @@ ClientSessionFactoryImpl.log.trace(this + "::Subscribing Topology"); } - channel0.send(new SubscribeClusterTopologyUpdatesMessageV2(serverLocator.isClusterConnection(), VersionLoader.getVersion().getIncrementingVersion())); + channel0.send(new SubscribeClusterTopologyUpdatesMessageV2(serverLocator.isClusterConnection(), + VersionLoader.getVersion() + .getIncrementingVersion())); } } @@ -1314,13 +1315,16 @@ /** * @param channel0 */ - public void sendNodeAnnounce(final long currentEventID, String nodeID, boolean isBackup, TransportConfiguration config, TransportConfiguration backupConfig) + public void sendNodeAnnounce(final long currentEventID, + String nodeID, + boolean isBackup, + TransportConfiguration config, + TransportConfiguration backupConfig) { Channel channel0 = connection.getChannel(0, -1); if (ClientSessionFactoryImpl.isDebug) { - ClientSessionFactoryImpl.log.debug("Announcing node " + serverLocator.getNodeID() + - ", isBackup=" + isBackup); + ClientSessionFactoryImpl.log.debug("Announcing node " + serverLocator.getNodeID() + ", isBackup=" + isBackup); } channel0.send(new NodeAnnounceMessage(currentEventID, nodeID, isBackup, config, backupConfig)); } @@ -1489,7 +1493,10 @@ " csf created at\nserverLocator=" + serverLocator, e); } - serverLocator.notifyNodeUp(System.currentTimeMillis(), topMessage.getNodeID(), topMessage.getPair(), topMessage.isLast()); + serverLocator.notifyNodeUp(System.currentTimeMillis(), + topMessage.getNodeID(), + topMessage.getPair(), + topMessage.isLast()); } } else if (type == PacketImpl.CLUSTER_TOPOLOGY_V2) @@ -1516,7 +1523,10 @@ " csf created at\nserverLocator=" + serverLocator, e); } - serverLocator.notifyNodeUp(topMessage.getUniqueEventID(), topMessage.getNodeID(), topMessage.getPair(), topMessage.isLast()); + serverLocator.notifyNodeUp(topMessage.getUniqueEventID(), + topMessage.getNodeID(), + topMessage.getPair(), + topMessage.isLast()); } } } @@ -1642,6 +1652,20 @@ } /* (non-Javadoc) + * @see java.lang.Object#toString() + */ + @Override + public String toString() + { + return "ClientSessionFactoryImpl [serverLocator=" + serverLocator + + ", connectorConfig=" + + connectorConfig + + ", backupConfig=" + + backupConfig + + "]"; + } + + /* (non-Javadoc) * @see org.hornetq.core.client.impl.ClientSessionFactoryInternal#setReconnectAttempts(int) */ public void setReconnectAttempts(final int attempts) Modified: branches/Branch_2_2_AS7/src/main/org/hornetq/core/client/impl/ClientSessionImpl.java =================================================================== --- branches/Branch_2_2_AS7/src/main/org/hornetq/core/client/impl/ClientSessionImpl.java 2011-11-07 04:32:51 UTC (rev 11665) +++ branches/Branch_2_2_AS7/src/main/org/hornetq/core/client/impl/ClientSessionImpl.java 2011-11-08 00:45:05 UTC (rev 11666) @@ -12,6 +12,8 @@ */ package org.hornetq.core.client.impl; +import java.security.AccessController; +import java.security.PrivilegedAction; import java.util.ArrayList; import java.util.HashMap; import java.util.HashSet; @@ -82,6 +84,7 @@ import org.hornetq.core.remoting.FailureListener; import org.hornetq.spi.core.protocol.RemotingConnection; import org.hornetq.spi.core.remoting.Connection; +import org.hornetq.spi.core.remoting.ConnectorFactory; import org.hornetq.utils.ConcurrentHashSet; import org.hornetq.utils.IDGenerator; import org.hornetq.utils.SimpleIDGenerator; @@ -925,7 +928,7 @@ { synchronized (this) { - if (closed) + if (closed/* || closing*/) { return; } @@ -1687,7 +1690,8 @@ username + ", closed=" + closed + - " metaData=(" + + ", factory = " + this.sessionFactory + + ", metaData=(" + buffer + ")]@" + Integer.toHexString(hashCode()); @@ -1778,7 +1782,8 @@ : null, executor, channel, - queueInfo); + queueInfo, + lookupTCCL()); addConsumer(consumer); @@ -1848,7 +1853,19 @@ throw new HornetQException(HornetQException.OBJECT_CLOSED, "Session is closed"); } } + + private ClassLoader lookupTCCL() + { + return AccessController.doPrivileged(new PrivilegedAction<ClassLoader>() + { + public ClassLoader run() + { + return Thread.currentThread().getContextClassLoader(); + } + }); + } + private void doCleanup(boolean failingOver) { if (remotingConnection == null) Modified: branches/Branch_2_2_AS7/src/main/org/hornetq/core/client/impl/ServerLocatorImpl.java =================================================================== --- branches/Branch_2_2_AS7/src/main/org/hornetq/core/client/impl/ServerLocatorImpl.java 2011-11-07 04:32:51 UTC (rev 11665) +++ branches/Branch_2_2_AS7/src/main/org/hornetq/core/client/impl/ServerLocatorImpl.java 2011-11-08 00:45:05 UTC (rev 11666) @@ -57,6 +57,15 @@ */ public class ServerLocatorImpl implements ServerLocatorInternal, DiscoveryListener, Serializable { + /*needed for backward compatibility*/ + private final Set<ClusterTopologyListener> topologyListeners = new HashSet<ClusterTopologyListener>(); + + /*end of compatibility fixes*/ + private enum STATE + { + INITIALIZED, CLOSED, CLOSING + }; + private static final long serialVersionUID = -1615857864410205260L; private static final Logger log = Logger.getLogger(ServerLocatorImpl.class); @@ -69,8 +78,10 @@ private transient String identity; - private Set<ClientSessionFactoryInternal> factories = new HashSet<ClientSessionFactoryInternal>(); + private final Set<ClientSessionFactoryInternal> factories = new HashSet<ClientSessionFactoryInternal>(); + private final Set<ClientSessionFactoryInternal> connectingFactories = new HashSet<ClientSessionFactoryInternal>(); + private TransportConfiguration[] initialConnectors; private DiscoveryGroupConfiguration discoveryGroupConfiguration; @@ -79,9 +90,9 @@ private final Topology topology; - private Pair<TransportConfiguration, TransportConfiguration>[] topologyArray; + private volatile Pair<TransportConfiguration, TransportConfiguration>[] topologyArray; - private boolean receivedTopology; + private volatile boolean receivedTopology; private boolean compressLargeMessage; @@ -154,10 +165,8 @@ private int initialMessagePacketSize; - private volatile boolean closed; + private volatile STATE state; - private volatile boolean closing; - private final List<Interceptor> interceptors = new CopyOnWriteArrayList<Interceptor>(); private static ExecutorService globalThreadPool; @@ -252,7 +261,7 @@ return globalScheduledThreadPool; } - private void setThreadPools() + private synchronized void setThreadPools() { if (threadPool != null) { @@ -329,10 +338,16 @@ }); } - private synchronized void initialise() throws Exception + private synchronized void initialise() throws HornetQException { - if (!readOnly) + if (readOnly) { + return; + } + + try + { + state = STATE.INITIALIZED; setThreadPools(); instantiateLoadBalancingPolicy(); @@ -366,6 +381,11 @@ readOnly = true; } + catch (Exception e) + { + state = null; + throw new HornetQException(HornetQException.INTERNAL_ERROR, "Failed to initialise session factory", e); + } } private ServerLocatorImpl(final Topology topology, @@ -375,7 +395,7 @@ { e.fillInStackTrace(); - this.topology = topology; + this.topology = topology == null ? new Topology(this) : topology; this.ha = useHA; @@ -454,7 +474,7 @@ */ public ServerLocatorImpl(final boolean useHA, final DiscoveryGroupConfiguration groupConfiguration) { - this(useHA ? new Topology(null) : null, useHA, groupConfiguration, null); + this(new Topology(null), useHA, groupConfiguration, null); if (useHA) { // We only set the owner at where the Topology was created. @@ -470,7 +490,7 @@ */ public ServerLocatorImpl(final boolean useHA, final TransportConfiguration... transportConfigs) { - this(useHA ? new Topology(null) : null, useHA, null, transportConfigs); + this(new Topology(null), useHA, null, transportConfigs); if (useHA) { // We only set the owner at where the Topology was created. @@ -505,7 +525,7 @@ this(topology, useHA, null, transportConfigs); } - private TransportConfiguration selectConnector() + private synchronized TransportConfiguration selectConnector() { if (receivedTopology) { @@ -515,14 +535,12 @@ return pair.getA(); } - else - { - // Get from initialconnectors - int pos = loadBalancingPolicy.select(initialConnectors.length); + // Get from initialconnectors - return initialConnectors[pos]; - } + int pos = loadBalancingPolicy.select(initialConnectors.length); + + return initialConnectors[pos]; } public void start(Executor executor) throws Exception @@ -541,7 +559,7 @@ } catch (Exception e) { - if (!closing) + if (!isClosed()) { log.warn("did not connect the cluster connection to other nodes", e); } @@ -565,19 +583,15 @@ public ClientSessionFactoryInternal connect() throws Exception { - ClientSessionFactoryInternal sf; // static list of initial connectors if (initialConnectors != null && discoveryGroup == null) { - sf = (ClientSessionFactoryInternal)staticConnector.connect(); + ClientSessionFactoryInternal sf = (ClientSessionFactoryInternal)staticConnector.connect(); + addFactory(sf); + return sf; } // wait for discovery group to get the list of initial connectors - else - { - sf = (ClientSessionFactoryInternal)createSessionFactory(); - } - addFactory(sf); - return sf; + return (ClientSessionFactoryInternal)createSessionFactory(); } /* (non-Javadoc) @@ -593,27 +607,45 @@ return afterConnectListener; } - public boolean isClosed() + public ClientSessionFactory createSessionFactory(String nodeID) throws Exception { - return closed || closing; - } + log.info(topology.describe("full topology")); + TopologyMember topologyMember = topology.getMember(nodeID); - public ClientSessionFactory createSessionFactory(final TransportConfiguration transportConfiguration) throws Exception - { - if (closed) + log.info("Creating connection factory towards " + nodeID + " = " + topologyMember); + + if (topologyMember == null) { - throw new IllegalStateException("Cannot create session factory, server locator is closed (maybe it has been garbage collected)"); + return null; } - - try + else if (topologyMember.getA() != null) { - initialise(); + ClientSessionFactoryInternal factory = (ClientSessionFactoryInternal)createSessionFactory(topologyMember.getA()); + if (topologyMember.getB() != null) + { + factory.setBackupConnector(topologyMember.getA(), topologyMember.getB()); + } + return factory; } - catch (Exception e) + else if (topologyMember.getA() == null && topologyMember.getB() != null) { - throw new HornetQException(HornetQException.INTERNAL_ERROR, "Failed to initialise session factory", e); + // This shouldn't happen, however I wanted this to consider all possible cases + ClientSessionFactoryInternal factory = (ClientSessionFactoryInternal)createSessionFactory(topologyMember.getB()); + return factory; } + else + { + // it shouldn't happen + return null; + } + } + public ClientSessionFactory createSessionFactory(final TransportConfiguration transportConfiguration) throws Exception + { + assertOpen(); + + initialise(); + ClientSessionFactoryInternal factory = new ClientSessionFactoryImpl(this, transportConfiguration, callTimeout, @@ -627,29 +659,39 @@ scheduledThreadPool, interceptors); - factory.connect(reconnectAttempts, failoverOnInitialConnection); + addToConnecting(factory); + try + { + factory.connect(reconnectAttempts, failoverOnInitialConnection); + addFactory(factory); + return factory; + } + finally + { + removeFromConnecting(factory); + } + } - addFactory(factory); - - return factory; + private void removeFromConnecting(ClientSessionFactoryInternal factory) + { + connectingFactories.remove(factory); } - public ClientSessionFactory createSessionFactory() throws Exception + private void addToConnecting(ClientSessionFactoryInternal factory) { - if (closed || closing) + synchronized (connectingFactories) { - throw new IllegalStateException("Cannot create session factory, server locator is closed (maybe it has been garbage collected)"); + assertOpen(); + connectingFactories.add(factory); } + } - try - { - initialise(); - } - catch (Exception e) - { - throw new HornetQException(HornetQException.INTERNAL_ERROR, "Failed to initialise session factory", e); - } + public ClientSessionFactory createSessionFactory() throws Exception + { + assertOpen(); + initialise(); + if (initialConnectors == null && discoveryGroup != null) { // Wait for an initial broadcast to give us at least one node in the cluster @@ -691,7 +733,15 @@ threadPool, scheduledThreadPool, interceptors); - factory.connect(initialConnectAttempts, failoverOnInitialConnection); + try + { + addToConnecting(factory); + factory.connect(initialConnectAttempts, failoverOnInitialConnection); + } + finally + { + removeFromConnecting(factory); + } } catch (HornetQException e) { @@ -723,10 +773,8 @@ if (ha || clusterConnection) { - long timeout = System.currentTimeMillis() + 30000; - while (!ServerLocatorImpl.this.closed && !ServerLocatorImpl.this.closing && - !receivedTopology && - timeout > System.currentTimeMillis()) + final long timeout = System.currentTimeMillis() + 30000; + while (!isClosed() && !receivedTopology && timeout > System.currentTimeMillis()) { // Now wait for the topology @@ -740,7 +788,7 @@ } - if (System.currentTimeMillis() > timeout && !receivedTopology && !closed && !closing) + if (System.currentTimeMillis() > timeout && !receivedTopology) { throw new HornetQException(HornetQException.CONNECTION_TIMEDOUT, "Timed out waiting to receive cluster topology. Group:" + discoveryGroup); @@ -1184,7 +1232,7 @@ protected void doClose(final boolean sendClose) { - if (closed) + if (state == STATE.CLOSED) { if (log.isDebugEnabled()) { @@ -1193,45 +1241,55 @@ return; } - if (log.isDebugEnabled()) - { - log.debug(this + " is calling close", new Exception("trace")); - } + state = STATE.CLOSING; - closing = true; - if (discoveryGroup != null) { - try + synchronized (this) { - discoveryGroup.stop(); + try + { + discoveryGroup.stop(); + } + catch (Exception e) + { + log.error("Failed to stop discovery group", e); + } } - catch (Exception e) - { - log.error("Failed to stop discovery group", e); - } } else { staticConnector.disconnect(); } - Set<ClientSessionFactoryInternal> clonedFactory = new HashSet<ClientSessionFactoryInternal>(factories); - - for (ClientSessionFactory factory : clonedFactory) + synchronized (connectingFactories) { - if (sendClose) + for (ClientSessionFactoryInternal csf : connectingFactories) { - factory.close(); + csf.close(); } - else + connectingFactories.clear(); + } + + synchronized (factories) + { + Set<ClientSessionFactoryInternal> clonedFactory = new HashSet<ClientSessionFactoryInternal>(factories); + + for (ClientSessionFactory factory : clonedFactory) { - factory.cleanup(); + if (sendClose) + { + factory.close(); + } + else + { + factory.cleanup(); + } } + + factories.clear(); } - factories.clear(); - if (shutdownPool) { if (threadPool != null) @@ -1268,7 +1326,8 @@ } readOnly = false; - closed = true; + state = STATE.CLOSED; + } /** This is directly called when the connection to the node is gone, @@ -1277,7 +1336,7 @@ public void notifyNodeDown(final long eventTime, final String nodeID) { - if (topology == null) + if (!ha) { // there's no topology here return; @@ -1288,25 +1347,31 @@ log.debug("nodeDown " + this + " nodeID=" + nodeID + " as being down", new Exception("trace")); } - if (topology.removeMember(eventTime, nodeID)) + topology.removeMember(eventTime, nodeID); + + if (clusterConnection) { - if (topology.isEmpty()) + updateArraysAndPairs(); + } + else + { + synchronized (this) { - // Resetting the topology to its original condition as it was brand new - synchronized (this) + if (topology.isEmpty()) { + // Resetting the topology to its original condition as it was brand new + receivedTopology = false; topologyArray = null; - receivedTopology = false; } - } - else - { - updateArraysAndPairs(); + else + { + updateArraysAndPairs(); - if (topology.nodes() == 1 && topology.getMember(this.nodeID) != null) - { - // Resetting the topology to its original condition as it was brand new - receivedTopology = false; + if (topology.nodes() == 1 && topology.getMember(this.nodeID) != null) + { + // Resetting the topology to its original condition as it was brand new + receivedTopology = false; + } } } } @@ -1318,7 +1383,7 @@ final Pair<TransportConfiguration, TransportConfiguration> connectorPair, final boolean last) { - if (topology == null) + if (!ha) { // there's no topology return; @@ -1331,23 +1396,21 @@ TopologyMember member = new TopologyMember(connectorPair.getA(), connectorPair.getB()); - if (topology.updateMember(uniqueEventID, nodeID, member)) - { + topology.updateMember(uniqueEventID, nodeID, member); - TopologyMember actMember = topology.getMember(nodeID); + TopologyMember actMember = topology.getMember(nodeID); - if (actMember != null && actMember.getConnector().getA() != null && actMember.getConnector().getB() != null) + if (actMember != null && actMember.getConnector().getA() != null && actMember.getConnector().getB() != null) + { + for (ClientSessionFactory factory : factories) { - for (ClientSessionFactory factory : factories) - { - ((ClientSessionFactoryInternal)factory).setBackupConnector(actMember.getConnector().getA(), - actMember.getConnector().getB()); - } + ((ClientSessionFactoryInternal)factory).setBackupConnector(actMember.getConnector().getA(), + actMember.getConnector().getB()); } - - updateArraysAndPairs(); } + updateArraysAndPairs(); + if (last) { synchronized (this) @@ -1374,13 +1437,10 @@ discoveryGroupConfiguration + "]"; } - else - { - return "ServerLocatorImpl [initialConnectors=" + Arrays.toString(initialConnectors) + - ", discoveryGroupConfiguration=" + - discoveryGroupConfiguration + - "]"; - } + return "ServerLocatorImpl [initialConnectors=" + Arrays.toString(initialConnectors) + + ", discoveryGroupConfiguration=" + + discoveryGroupConfiguration + + "]"; } private synchronized void updateArraysAndPairs() @@ -1409,7 +1469,7 @@ { this.initialConnectors[count++] = entry.getConnector(); - if (topology != null && topology.getMember(entry.getNodeID()) == null) + if (ha && topology.getMember(entry.getNodeID()) == null) { TopologyMember member = new TopologyMember(entry.getConnector(), null); // on this case we set it as zero as any update coming from server should be accepted @@ -1461,13 +1521,24 @@ topology.removeClusterTopologyListener(listener); } - public synchronized void addFactory(ClientSessionFactoryInternal factory) + private synchronized void addFactory(ClientSessionFactoryInternal factory) { - if (factory != null) + if (factory == null) { + return; + } + + synchronized (factories) + { + if (isClosed()) + { + factory.close(); + return; + } + TransportConfiguration backup = null; - if (topology != null) + if (ha) { backup = topology.getBackupForConnector(factory.getConnectorConfiguration()); } @@ -1485,19 +1556,9 @@ public ClientSessionFactory connect() throws HornetQException { - if (closed) - { - throw new IllegalStateException("Cannot create session factory, server locator is closed (maybe it has been garbage collected)"); - } + assertOpen(); - try - { - initialise(); - } - catch (Exception e) - { - throw new HornetQException(HornetQException.INTERNAL_ERROR, "Failed to initialise session factory", e); - } + initialise(); ClientSessionFactory csf = null; @@ -1507,7 +1568,7 @@ { int retryNumber = 0; - while (csf == null && !ServerLocatorImpl.this.closed && !ServerLocatorImpl.this.closing) + while (csf == null && !isClosed()) { retryNumber++; for (Connector conn : connectors) @@ -1560,7 +1621,7 @@ break; } - if (!closed && !closing) + if (!isClosed()) { Thread.sleep(retryInterval); } @@ -1573,7 +1634,7 @@ throw new HornetQException(HornetQException.NOT_CONNECTED, "Failed to connect to any static connectors", e); } - if (csf == null && !closed) + if (csf == null && !isClosed()) { log.warn("Failed to connecto to any static connector, throwing exception now"); throw new HornetQException(HornetQException.NOT_CONNECTED, "Failed to connect to any static connectors"); @@ -1632,7 +1693,7 @@ public void finalize() throws Throwable { - if (!closed && finalizeCheck) + if (!isClosed() && finalizeCheck) { log.warn("I'm closing a core ServerLocator you left open. Please make sure you close all ServerLocators explicitly " + "before letting them go out of scope! " + System.identityHashCode(this)); @@ -1677,7 +1738,15 @@ ClientSessionFactoryInternal factoryToUse = factory; if (factoryToUse != null) { - factory.connect(1, false); + try + { + addToConnecting(factoryToUse); + factoryToUse.connect(1, false); + } + finally + { + removeFromConnecting(factoryToUse); + } } return factoryToUse; } @@ -1700,9 +1769,6 @@ } } - /* (non-Javadoc) - * @see java.lang.Object#toString() - */ @Override public String toString() { @@ -1711,4 +1777,17 @@ } } + + private void assertOpen() + { + if (state != null && state != STATE.INITIALIZED) + { + throw new IllegalStateException("Cannot create session factory, server locator is closed (maybe it has been garbage collected)"); + } + } + + public boolean isClosed() + { + return state != STATE.INITIALIZED; + } } Modified: branches/Branch_2_2_AS7/src/main/org/hornetq/core/client/impl/Topology.java =================================================================== --- branches/Branch_2_2_AS7/src/main/org/hornetq/core/client/impl/Topology.java 2011-11-07 04:32:51 UTC (rev 11665) +++ branches/Branch_2_2_AS7/src/main/org/hornetq/core/client/impl/Topology.java 2011-11-08 00:45:05 UTC (rev 11666) @@ -58,9 +58,9 @@ * keys are node IDs * values are a pair of live/backup transport configurations */ - private final Map<String, TopologyMember> mapTopology = new ConcurrentHashMap<String, TopologyMember>(); + private final Map<String, TopologyMember> topology = new ConcurrentHashMap<String, TopologyMember>(); - private final Map<String, Long> mapDelete = new ConcurrentHashMap<String, Long>(); + private transient final Map<String, Long> mapDelete = new ConcurrentHashMap<String, Long>(); public Topology(final Object owner) { @@ -111,8 +111,8 @@ log.debug(this + "::node " + nodeId + "=" + memberInput); } memberInput.setUniqueEventID(System.currentTimeMillis()); - mapTopology.remove(nodeId); - mapTopology.put(nodeId, memberInput); + topology.remove(nodeId); + topology.put(nodeId, memberInput); sendMemberUp(memberInput.getUniqueEventID(), nodeId, memberInput); } } @@ -130,17 +130,17 @@ TopologyMember currentMember = getMember(nodeId); if (currentMember == null) { - log.warn("There's no live to be updated on backup update, node=" + nodeId + " memberInput=" + memberInput, + log.debug("There's no live to be updated on backup update, node=" + nodeId + " memberInput=" + memberInput, new Exception("trace")); currentMember = memberInput; - mapTopology.put(nodeId, currentMember); + topology.put(nodeId, currentMember); } TopologyMember newMember = new TopologyMember(currentMember.getA(), memberInput.getB()); newMember.setUniqueEventID(System.currentTimeMillis()); - mapTopology.remove(nodeId); - mapTopology.put(nodeId, newMember); + topology.remove(nodeId); + topology.put(nodeId, newMember); sendMemberUp(newMember.getUniqueEventID(), nodeId, newMember); return newMember; @@ -172,7 +172,7 @@ synchronized (this) { - TopologyMember currentMember = mapTopology.get(nodeId); + TopologyMember currentMember = topology.get(nodeId); if (currentMember == null) { @@ -184,7 +184,7 @@ memberInput, new Exception("trace")); } memberInput.setUniqueEventID(uniqueEventID); - mapTopology.put(nodeId, memberInput); + topology.put(nodeId, memberInput); sendMemberUp(uniqueEventID, nodeId, memberInput); return true; } @@ -217,8 +217,8 @@ newMember.setUniqueEventID(uniqueEventID); - mapTopology.remove(nodeId); - mapTopology.put(nodeId, newMember); + topology.remove(nodeId); + topology.put(nodeId, newMember); sendMemberUp(uniqueEventID, nodeId, newMember); return true; @@ -296,7 +296,7 @@ synchronized (this) { - member = mapTopology.get(nodeId); + member = topology.get(nodeId); if (member != null) { if (member.getUniqueEventID() > uniqueEventID) @@ -307,7 +307,7 @@ else { mapDelete.put(nodeId, uniqueEventID); - member = mapTopology.remove(nodeId); + member = topology.remove(nodeId); } } } @@ -320,7 +320,7 @@ ", result=" + member + ", size = " + - mapTopology.size(), new Exception("trace")); + topology.size(), new Exception("trace")); } if (member != null) @@ -414,7 +414,7 @@ synchronized (Topology.this) { - copy = new HashMap<String, TopologyMember>(mapTopology); + copy = new HashMap<String, TopologyMember>(topology); } for (Map.Entry<String, TopologyMember> entry : copy.entrySet()) @@ -439,12 +439,12 @@ public synchronized TopologyMember getMember(final String nodeID) { - return mapTopology.get(nodeID); + return topology.get(nodeID); } public synchronized boolean isEmpty() { - return mapTopology.isEmpty(); + return topology.isEmpty(); } public Collection<TopologyMember> getMembers() @@ -452,7 +452,7 @@ ArrayList<TopologyMember> members; synchronized (this) { - members = new ArrayList<TopologyMember>(mapTopology.values()); + members = new ArrayList<TopologyMember>(topology.values()); } return members; } @@ -460,7 +460,7 @@ public synchronized int nodes() { int count = 0; - for (TopologyMember member : mapTopology.values()) + for (TopologyMember member : topology.values()) { if (member.getA() != null) { @@ -483,12 +483,12 @@ { String desc = text + "topology on " + this + ":\n"; - for (Entry<String, TopologyMember> entry : new HashMap<String, TopologyMember>(mapTopology).entrySet()) + for (Entry<String, TopologyMember> entry : new HashMap<String, TopologyMember>(topology).entrySet()) { desc += "\t" + entry.getKey() + " => " + entry.getValue() + "\n"; } desc += "\t" + "nodes=" + nodes() + "\t" + "members=" + members(); - if (mapTopology.isEmpty()) + if (topology.isEmpty()) { desc += "\tEmpty"; } @@ -497,7 +497,7 @@ public int members() { - return mapTopology.size(); + return topology.size(); } /** The owner exists mainly for debug purposes. @@ -510,7 +510,7 @@ public TransportConfiguration getBackupForConnector(final TransportConfiguration connectorConfiguration) { - for (TopologyMember member : mapTopology.values()) + for (TopologyMember member : topology.values()) { if (member.getA() != null && member.getA().equals(connectorConfiguration)) { Modified: branches/Branch_2_2_AS7/src/main/org/hornetq/core/config/BridgeConfiguration.java =================================================================== --- branches/Branch_2_2_AS7/src/main/org/hornetq/core/config/BridgeConfiguration.java 2011-11-07 04:32:51 UTC (rev 11665) +++ branches/Branch_2_2_AS7/src/main/org/hornetq/core/config/BridgeConfiguration.java 2011-11-08 00:45:05 UTC (rev 11666) @@ -64,6 +64,8 @@ private final long connectionTTL; private final long maxRetryInterval; + + private final int minLargeMessageSize; public BridgeConfiguration(final String name, @@ -71,6 +73,7 @@ final String forwardingAddress, final String filterString, final String transformerClassName, + final int minLargeMessageSize, final long clientFailureCheckPeriod, final long connectionTTL, final long retryInterval, @@ -87,6 +90,7 @@ this.name = name; this.queueName = queueName; this.forwardingAddress = forwardingAddress; + this.minLargeMessageSize = minLargeMessageSize; this.filterString = filterString; this.transformerClassName = transformerClassName; this.retryInterval = retryInterval; @@ -108,6 +112,7 @@ final String forwardingAddress, final String filterString, final String transformerClassName, + final int minLargeMessageSize, final long clientFailureCheckPeriod, final long connectionTTL, final long retryInterval, @@ -126,6 +131,7 @@ this.forwardingAddress = forwardingAddress; this.filterString = filterString; this.transformerClassName = transformerClassName; + this.minLargeMessageSize = minLargeMessageSize; this.retryInterval = retryInterval; this.retryIntervalMultiplier = retryIntervalMultiplier; this.reconnectAttempts = reconnectAttempts; @@ -244,6 +250,14 @@ } /** + * @return the minLargeMessageSize + */ + public int getMinLargeMessageSize() + { + return minLargeMessageSize; + } + + /** * @param forwardingAddress the forwardingAddress to set */ public void setForwardingAddress(final String forwardingAddress) Modified: branches/Branch_2_2_AS7/src/main/org/hornetq/core/config/ClusterConnectionConfiguration.java =================================================================== --- branches/Branch_2_2_AS7/src/main/org/hornetq/core/config/ClusterConnectionConfiguration.java 2011-11-07 04:32:51 UTC (rev 11665) +++ branches/Branch_2_2_AS7/src/main/org/hornetq/core/config/ClusterConnectionConfiguration.java 2011-11-08 00:45:05 UTC (rev 11666) @@ -66,6 +66,8 @@ private final boolean allowDirectConnectionsOnly; + private final int minLargeMessageSize; + public ClusterConnectionConfiguration(final String name, final String address, final String connectorName, @@ -80,6 +82,7 @@ this(name, address, connectorName, + HornetQClient.DEFAULT_MIN_LARGE_MESSAGE_SIZE, ConfigurationImpl.DEFAULT_CLUSTER_FAILURE_CHECK_PERIOD, ConfigurationImpl.DEFAULT_CLUSTER_CONNECTION_TTL, retryInterval, @@ -99,6 +102,7 @@ public ClusterConnectionConfiguration(final String name, final String address, final String connectorName, + final int minLargeMessageSize, final long clientFailureCheckPeriod, final long connectionTTL, final long retryInterval, @@ -130,6 +134,7 @@ this.maxHops = maxHops; this.confirmationWindowSize = confirmationWindowSize; this.allowDirectConnectionsOnly = allowDirectConnectionsOnly; + this.minLargeMessageSize = minLargeMessageSize; } @@ -146,6 +151,7 @@ this(name, address, connectorName, + HornetQClient.DEFAULT_MIN_LARGE_MESSAGE_SIZE, ConfigurationImpl.DEFAULT_CLUSTER_FAILURE_CHECK_PERIOD, ConfigurationImpl.DEFAULT_CLUSTER_CONNECTION_TTL, retryInterval, @@ -164,6 +170,7 @@ public ClusterConnectionConfiguration(final String name, final String address, final String connectorName, + final int minLargeMessageSize, final long clientFailureCheckPeriod, final long connectionTTL, final long retryInterval, @@ -193,6 +200,7 @@ this.staticConnectors = null; this.maxHops = maxHops; this.confirmationWindowSize = confirmationWindowSize; + this.minLargeMessageSize = minLargeMessageSize; allowDirectConnectionsOnly = false; } @@ -295,4 +303,15 @@ { return allowDirectConnectionsOnly; } + + + /** + * @return the minLargeMessageSize + */ + public int getMinLargeMessageSize() + { + return minLargeMessageSize; + } + + } Modified: branches/Branch_2_2_AS7/src/main/org/hornetq/core/deployers/impl/FileConfigurationParser.java =================================================================== --- branches/Branch_2_2_AS7/src/main/org/hornetq/core/deployers/impl/FileConfigurationParser.java 2011-11-07 04:32:51 UTC (rev 11665) +++ branches/Branch_2_2_AS7/src/main/org/hornetq/core/deployers/impl/FileConfigurationParser.java 2011-11-08 00:45:05 UTC (rev 11666) @@ -1021,6 +1021,8 @@ double retryIntervalMultiplier = XMLConfigurationUtil.getDouble(e, "retry-interval-multiplier", ConfigurationImpl.DEFAULT_CLUSTER_RETRY_INTERVAL_MULTIPLIER, Validators.GT_ZERO); + int minLargeMessageSize = XMLConfigurationUtil.getInteger(e, "min-large-message-size", HornetQClient.DEFAULT_MIN_LARGE_MESSAGE_SIZE, Validators.GT_ZERO); + long maxRetryInterval = XMLConfigurationUtil.getLong(e, "max-retry-interval", ConfigurationImpl.DEFAULT_CLUSTER_MAX_RETRY_INTERVAL, Validators.GT_ZERO); int reconnectAttempts = XMLConfigurationUtil.getInteger(e, "reconnect-attempts", ConfigurationImpl.DEFAULT_CLUSTER_RECONNECT_ATTEMPTS, Validators.MINUS_ONE_OR_GE_ZERO); @@ -1065,6 +1067,7 @@ config = new ClusterConnectionConfiguration(name, address, connectorName, + minLargeMessageSize, clientFailureCheckPeriod, connectionTTL, retryInterval, @@ -1084,6 +1087,7 @@ config = new ClusterConnectionConfiguration(name, address, connectorName, + minLargeMessageSize, clientFailureCheckPeriod, connectionTTL, retryInterval, @@ -1147,6 +1151,11 @@ long connectionTTL = XMLConfigurationUtil.getLong(brNode, "connection-ttl", HornetQClient.DEFAULT_CONNECTION_TTL, Validators.GT_ZERO) ; + int minLargeMessageSize = XMLConfigurationUtil.getInteger(brNode, + "min-large-message-size", + HornetQClient.DEFAULT_MIN_LARGE_MESSAGE_SIZE, + Validators.GT_ZERO); + long maxRetryInterval = XMLConfigurationUtil.getLong(brNode, "max-retry-interval", HornetQClient.DEFAULT_MAX_RETRY_INTERVAL, Validators.GT_ZERO); @@ -1211,6 +1220,7 @@ forwardingAddress, filterString, transformerClassName, + minLargeMessageSize, clientFailureCheckPeriod, connectionTTL, retryInterval, @@ -1231,6 +1241,7 @@ forwardingAddress, filterString, transformerClassName, + minLargeMessageSize, clientFailureCheckPeriod, connectionTTL, retryInterval, Modified: branches/Branch_2_2_AS7/src/main/org/hornetq/core/journal/SequentialFile.java =================================================================== --- branches/Branch_2_2_AS7/src/main/org/hornetq/core/journal/SequentialFile.java 2011-11-07 04:32:51 UTC (rev 11665) +++ branches/Branch_2_2_AS7/src/main/org/hornetq/core/journal/SequentialFile.java 2011-11-08 00:45:05 UTC (rev 11666) @@ -93,7 +93,7 @@ void renameTo(String newFileName) throws Exception; - SequentialFile copy(); + SequentialFile cloneFile(); void copyTo(SequentialFile newFileName) throws Exception; Modified: branches/Branch_2_2_AS7/src/main/org/hornetq/core/journal/impl/AIOSequentialFile.java =================================================================== --- branches/Branch_2_2_AS7/src/main/org/hornetq/core/journal/impl/AIOSequentialFile.java 2011-11-07 04:32:51 UTC (rev 11665) +++ branches/Branch_2_2_AS7/src/main/org/hornetq/core/journal/impl/AIOSequentialFile.java 2011-11-08 00:45:05 UTC (rev 11666) @@ -86,7 +86,7 @@ return pos; } - public SequentialFile copy() + public SequentialFile cloneFile() { return new AIOSequentialFile(factory, -1, Modified: branches/Branch_2_2_AS7/src/main/org/hornetq/core/journal/impl/AbstractSequentialFile.java =================================================================== --- branches/Branch_2_2_AS7/src/main/org/hornetq/core/journal/impl/AbstractSequentialFile.java 2011-11-07 04:32:51 UTC (rev 11665) +++ branches/Branch_2_2_AS7/src/main/org/hornetq/core/journal/impl/AbstractSequentialFile.java 2011-11-08 00:45:05 UTC (rev 11666) @@ -109,7 +109,10 @@ { log.debug("Copying " + this + " as " + newFileName); newFileName.open(); - this.open(); + if (!isOpen()) + { + this.open(); + } ByteBuffer buffer = ByteBuffer.allocate(10 * 1024); Modified: branches/Branch_2_2_AS7/src/main/org/hornetq/core/journal/impl/NIOSequentialFile.java =================================================================== --- branches/Branch_2_2_AS7/src/main/org/hornetq/core/journal/impl/NIOSequentialFile.java 2011-11-07 04:32:51 UTC (rev 11665) +++ branches/Branch_2_2_AS7/src/main/org/hornetq/core/journal/impl/NIOSequentialFile.java 2011-11-08 00:45:05 UTC (rev 11666) @@ -252,7 +252,7 @@ return "NIOSequentialFile " + getFile(); } - public SequentialFile copy() + public SequentialFile cloneFile() { return new NIOSequentialFile(factory, getFile(), maxIO, writerExecutor); } Modified: branches/Branch_2_2_AS7/src/main/org/hornetq/core/management/impl/HornetQServerControlImpl.java =================================================================== --- branches/Branch_2_2_AS7/src/main/org/hornetq/core/management/impl/HornetQServerControlImpl.java 2011-11-07 04:32:51 UTC (rev 11665) +++ branches/Branch_2_2_AS7/src/main/org/hornetq/core/management/impl/HornetQServerControlImpl.java 2011-11-08 00:45:05 UTC (rev 11666) @@ -1709,6 +1709,7 @@ forwardingAddress, filterString, transformerClassName, + HornetQClient.DEFAULT_MIN_LARGE_MESSAGE_SIZE, clientFailureCheckPeriod, HornetQClient.DEFAULT_CONNECTION_TTL, retryInterval, @@ -1730,6 +1731,7 @@ forwardingAddress, filterString, transformerClassName, + HornetQClient.DEFAULT_MIN_LARGE_MESSAGE_SIZE, clientFailureCheckPeriod, HornetQClient.DEFAULT_CONNECTION_TTL, retryInterval, Modified: branches/Branch_2_2_AS7/src/main/org/hornetq/core/paging/impl/PageImpl.java =================================================================== --- branches/Branch_2_2_AS7/src/main/org/hornetq/core/paging/impl/PageImpl.java 2011-11-07 04:32:51 UTC (rev 11665) +++ branches/Branch_2_2_AS7/src/main/org/hornetq/core/paging/impl/PageImpl.java 2011-11-08 00:45:05 UTC (rev 11666) @@ -212,7 +212,10 @@ public void open() throws Exception { - file.open(); + if (!file.isOpen()) + { + file.open(); + } size.set((int)file.size()); file.position(0); } @@ -307,6 +310,21 @@ { return otherPage.getPageId() - this.pageId; } + + public void finalize() + { + try + { + if (file != null && file.isOpen()) + { + file.close(); + } + } + catch (Exception e) + { + log.warn(e.getMessage(), e); + } + } /* (non-Javadoc) Modified: branches/Branch_2_2_AS7/src/main/org/hornetq/core/paging/impl/PagingManagerImpl.java =================================================================== --- branches/Branch_2_2_AS7/src/main/org/hornetq/core/paging/impl/PagingManagerImpl.java 2011-11-07 04:32:51 UTC (rev 11665) +++ branches/Branch_2_2_AS7/src/main/org/hornetq/core/paging/impl/PagingManagerImpl.java 2011-11-08 00:45:05 UTC (rev 11666) @@ -114,8 +114,17 @@ for (PagingStore store : reloadedStores) { - store.start(); - stores.put(store.getStoreName(), store); + PagingStore oldStore = stores.get(store.getStoreName()); + if (oldStore != null) + { + oldStore.stop(); + oldStore.start(); + } + else + { + store.start(); + stores.put(store.getStoreName(), store); + } } } Modified: branches/Branch_2_2_AS7/src/main/org/hornetq/core/paging/impl/PagingStoreImpl.java =================================================================== --- branches/Branch_2_2_AS7/src/main/org/hornetq/core/paging/impl/PagingStoreImpl.java 2011-11-07 04:32:51 UTC (rev 11665) +++ branches/Branch_2_2_AS7/src/main/org/hornetq/core/paging/impl/PagingStoreImpl.java 2011-11-08 00:45:05 UTC (rev 11666) @@ -426,6 +426,10 @@ { currentPageId = 0; + if (currentPage != null) + { + currentPage.close(); + } currentPage = null; List<String> files = fileFactory.listFiles("page"); @@ -815,6 +819,11 @@ PagingStoreImpl.log.warn("Messages are being dropped on address " + getStoreName()); } + + if (log.isDebugEnabled()) + { + log.debug("Message " + message + " beig dropped for fullAddressPolicy==DROP"); + } // Address is full, we just pretend we are paging, and drop the data return true; Modified: branches/Branch_2_2_AS7/src/main/org/hornetq/core/persistence/impl/journal/JournalStorageManager.java =================================================================== --- branches/Branch_2_2_AS7/src/main/org/hornetq/core/persistence/impl/journal/JournalStorageManager.java 2011-11-07 04:32:51 UTC (rev 11665) +++ branches/Branch_2_2_AS7/src/main/org/hornetq/core/persistence/impl/journal/JournalStorageManager.java 2011-11-08 00:45:05 UTC (rev 11666) @@ -1275,11 +1275,11 @@ log.info("Deleting unreferenced message id=" + msg.getMessageID() + " from the journal"); try { - deleteMessage(msg.getMessageID()); + deleteMessage(msg.getMessageID()); } catch (Exception ignored) { - log.warn("It wasn't possible to delete message " + msg.getMessageID()); + log.warn("It wasn't possible to delete message " + msg.getMessageID(), ignored); } } } @@ -1985,7 +1985,7 @@ } } - private OperationContext getContext(final boolean sync) + protected OperationContext getContext(final boolean sync) { if (sync) { Modified: branches/Branch_2_2_AS7/src/main/org/hornetq/core/persistence/impl/journal/LargeServerMessageImpl.java =================================================================== --- branches/Branch_2_2_AS7/src/main/org/hornetq/core/persistence/impl/journal/LargeServerMessageImpl.java 2011-11-07 04:32:51 UTC (rev 11665) +++ branches/Branch_2_2_AS7/src/main/org/hornetq/core/persistence/impl/journal/LargeServerMessageImpl.java 2011-11-08 00:45:05 UTC (rev 11666) @@ -170,11 +170,21 @@ public synchronized void incrementDelayDeletionCount() { delayDeletionCount.incrementAndGet(); + try + { + incrementRefCount(); + } + catch (Exception e) + { + log.warn(e.getMessage(), e); + } } public synchronized void decrementDelayDeletionCount() throws Exception { int count = delayDeletionCount.decrementAndGet(); + + decrementRefCount(); if (count == 0) { @@ -373,6 +383,10 @@ log.warn("Error on copying large message this for DLA or Expiry", e); return null; } + finally + { + releaseResources(); + } } } @@ -416,7 +430,7 @@ file = storageManager.createFileForLargeMessage(getMessageID(), durable); - file.open(); + openFile(); bodySize = file.size(); } @@ -427,6 +441,27 @@ throw new HornetQException(HornetQException.INTERNAL_ERROR, e.getMessage(), e); } } + + protected void openFile() throws Exception + { + if (file == null) + { + validateFile(); + } + else + if (!file.isOpen()) + { + file.open(); + } + } + + protected void closeFile() throws Exception + { + if (file != null && file.isOpen()) + { + file.close(); + } + } /* (non-Javadoc) * @see org.hornetq.core.server.LargeServerMessage#setLinkedMessage(org.hornetq.core.server.LargeServerMessage) @@ -444,9 +479,9 @@ file = storageManager.createFileForLargeMessage(message.getMessageID(), durable); try { - file.open(); + openFile(); bodySize = file.size(); - file.close(); + closeFile(); } catch (Exception e) { @@ -464,7 +499,11 @@ { try { - cFile = file.copy(); + if (cFile != null && cFile.isOpen()) + { + cFile.close(); + } + cFile = file.cloneFile(); cFile.open(); } catch (Exception e) Modified: branches/Branch_2_2_AS7/src/main/org/hornetq/core/persistence/impl/journal/PrintData.java =================================================================== --- branches/Branch_2_2_AS7/src/main/org/hornetq/core/persistence/impl/journal/PrintData.java 2011-11-07 04:32:51 UTC (rev 11665) +++ branches/Branch_2_2_AS7/src/main/org/hornetq/core/persistence/impl/journal/PrintData.java 2011-11-08 00:45:05 UTC (rev 11666) @@ -13,6 +13,9 @@ package org.hornetq.core.persistence.impl.journal; +import java.io.File; + +import org.hornetq.core.server.impl.FileLockNodeManager; /** * A PrintData * @@ -42,6 +45,25 @@ System.exit(-1); } + File serverLockFile = new File(arg[1], "server.lock"); + + if (serverLockFile.isFile()) + { + try + { + FileLockNodeManager fileLock = new FileLockNodeManager(arg[1]); + fileLock.start(); + System.out.println("********************************************"); + System.out.println("Server's ID=" + fileLock.getNodeId().toString()); + System.out.println("********************************************"); + fileLock.stop(); + } + catch (Exception e) + { + e.printStackTrace(); + } + } + System.out.println("********************************************"); System.out.println("B I N D I N G S J O U R N A L"); System.out.println("********************************************"); Modified: branches/Branch_2_2_AS7/src/main/org/hornetq/core/postoffice/impl/BindingsImpl.java =================================================================== --- branches/Branch_2_2_AS7/src/main/org/hornetq/core/postoffice/impl/BindingsImpl.java 2011-11-07 04:32:51 UTC (rev 11665) +++ branches/Branch_2_2_AS7/src/main/org/hornetq/core/postoffice/impl/BindingsImpl.java 2011-11-08 00:45:05 UTC (rev 11666) @@ -595,6 +595,10 @@ { binding.route(message, context); } + else + { + log.warn("Couldn't find binding with id=" + bindingID + " on routeFromCluster for message=" + message + " binding = " + this); + } } } Modified: branches/Branch_2_2_AS7/src/main/org/hornetq/core/postoffice/impl/PostOfficeImpl.java =================================================================== --- branches/Branch_2_2_AS7/src/main/org/hornetq/core/postoffice/impl/PostOfficeImpl.java 2011-11-07 04:32:51 UTC (rev 11665) +++ branches/Branch_2_2_AS7/src/main/org/hornetq/core/postoffice/impl/PostOfficeImpl.java 2011-11-08 00:45:05 UTC (rev 11666) @@ -473,9 +473,9 @@ String uid = UUIDGenerator.getInstance().generateStringUUID(); - if (isTrace) + if (log.isDebugEnabled()) { - log.trace("Sending notification for addBinding " + binding + " from server " + server); + log.debug("ClusterCommunication::Sending notification for addBinding " + binding + " from server " + server); } managementService.sendNotification(new Notification(uid, NotificationType.BINDING_ADDED, props)); @@ -611,6 +611,19 @@ { bindings.route(message, context); } + else + { + // this is a debug and not warn because this could be a regular scenario on publish-subscribe queues (or topic subscriptions on JMS) + if (log.isDebugEnabled()) + { + log.debug("Couldn't find any bindings for address=" + address + " on message=" + message); + } + } + + if (log.isTraceEnabled()) + { + log.trace("Message after routed=" + message); + } if (context.getQueueCount() == 0) { @@ -625,6 +638,11 @@ // Send to the DLA for the address SimpleString dlaAddress = addressSettings.getDeadLetterAddress(); + + if (log.isDebugEnabled()) + { + log.debug("sending message to dla address = " + dlaAddress + ", message=" + message); + } if (dlaAddress == null) { @@ -641,6 +659,13 @@ route(message, context.getTransaction(), false); } } + else + { + if (log.isDebugEnabled()) + { + log.debug("Message " + message + " is not going anywhere as it didn't have a binding on address:" + address); + } + } } else { Modified: branches/Branch_2_2_AS7/src/main/org/hornetq/core/postoffice/impl/SimpleAddressManager.java =================================================================== --- branches/Branch_2_2_AS7/src/main/org/hornetq/core/postoffice/impl/SimpleAddressManager.java 2011-11-07 04:32:51 UTC (rev 11665) +++ branches/Branch_2_2_AS7/src/main/org/hornetq/core/postoffice/impl/SimpleAddressManager.java 2011-11-08 00:45:05 UTC (rev 11666) @@ -52,6 +52,11 @@ { throw new IllegalStateException("Binding already exists " + binding); } + + if (log.isDebugEnabled()) + { + log.debug("Adding binding " + binding + " with address = " + binding.getUniqueName(), new Exception ("trace")); + } return addMappingInternal(binding.getAddress(), binding); } Modified: branches/Branch_2_2_AS7/src/main/org/hornetq/core/protocol/core/ServerSessionPacketHandler.java =================================================================== --- branches/Branch_2_2_AS7/src/main/org/hornetq/core/protocol/core/ServerSessionPacketHandler.java 2011-11-07 04:32:51 UTC (rev 11665) +++ branches/Branch_2_2_AS7/src/main/org/hornetq/core/protocol/core/ServerSessionPacketHandler.java 2011-11-08 00:45:05 UTC (rev 11666) @@ -514,6 +514,7 @@ { if (requiresResponse) { + log.debug("Sending exception to client", e); response = new SessionXAResponseMessage(true, e.errorCode, e.getMessage()); } else @@ -525,6 +526,7 @@ { if (requiresResponse) { + log.debug("Sending exception to client", e); response = new HornetQExceptionMessage((HornetQException)e); } else Modified: branches/Branch_2_2_AS7/src/main/org/hornetq/core/protocol/core/impl/CoreProtocolManager.java =================================================================== --- branches/Branch_2_2_AS7/src/main/org/hornetq/core/protocol/core/impl/CoreProtocolManager.java 2011-11-07 04:32:51 UTC (rev 11665) +++ branches/Branch_2_2_AS7/src/main/org/hornetq/core/protocol/core/impl/CoreProtocolManager.java 2011-11-08 00:45:05 UTC (rev 11666) @@ -39,6 +39,7 @@ import org.hornetq.core.protocol.core.impl.wireformat.SubscribeClusterTopologyUpdatesMessageV2; import org.hornetq.core.remoting.CloseListener; import org.hornetq.core.server.HornetQServer; +import org.hornetq.core.server.cluster.ClusterConnection; import org.hornetq.spi.core.protocol.ConnectionEntry; import org.hornetq.spi.core.protocol.ProtocolManager; import org.hornetq.spi.core.protocol.RemotingConnection; @@ -208,8 +209,23 @@ { log.trace("Server " + server + " receiving nodeUp from NodeID=" + msg.getNodeID() + ", pair=" + pair); } - - acceptorUsed.getClusterConnection().nodeAnnounced(msg.getCurrentEventID(), msg.getNodeID(), pair, msg.isBackup()); + + if (acceptorUsed != null) + { + ClusterConnection clusterConn = acceptorUsed.getClusterConnection(); + if (clusterConn != null) + { + clusterConn.nodeAnnounced(msg.getCurrentEventID(), msg.getNodeID(), pair, msg.isBackup()); + } + else + { + log.debug("Cluster connection is null on acceptor = " + acceptorUsed); + } + } + else + { + log.debug("there is no acceptor used configured at the CoreProtocolManager " + this); + } } } }); Modified: branches/Branch_2_2_AS7/src/main/org/hornetq/core/remoting/impl/netty/NettyConnection.java =================================================================== --- branches/Branch_2_2_AS7/src/main/org/hornetq/core/remoting/impl/netty/NettyConnection.java 2011-11-07 04:32:51 UTC (rev 11665) +++ branches/Branch_2_2_AS7/src/main/org/hornetq/core/remoting/impl/netty/NettyConnection.java 2011-11-08 00:45:05 UTC (rev 11666) @@ -14,7 +14,7 @@ package org.hornetq.core.remoting.impl.netty; import java.util.Set; -import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.Semaphore; import org.hornetq.api.core.HornetQBuffer; import org.hornetq.api.core.HornetQBuffers; @@ -60,27 +60,27 @@ private volatile HornetQBuffer batchBuffer; - private final AtomicBoolean writeLock = new AtomicBoolean(false); - - private Set<ReadyListener> readyListeners = new ConcurrentHashSet<ReadyListener>(); + private final Semaphore writeLock = new Semaphore(1); + private final Set<ReadyListener> readyListeners = new ConcurrentHashSet<ReadyListener>(); + // Static -------------------------------------------------------- // Constructors -------------------------------------------------- public NettyConnection(final Channel channel, final ConnectionLifeCycleListener listener, - boolean batchingEnabled, - boolean directDeliver) + final boolean batchingEnabled, + final boolean directDeliver) { this(null, channel, listener, batchingEnabled, directDeliver); } - + public NettyConnection(final Acceptor acceptor, final Channel channel, final ConnectionLifeCycleListener listener, - boolean batchingEnabled, - boolean directDeliver) + final boolean batchingEnabled, + final boolean directDeliver) { this.channel = channel; @@ -152,7 +152,7 @@ return; } - if (writeLock.compareAndSet(false, true)) + if (writeLock.tryAcquire()) { try { @@ -160,12 +160,12 @@ { channel.write(batchBuffer.channelBuffer()); - batchBuffer = HornetQBuffers.dynamicBuffer(BATCHING_BUFFER_SIZE); + batchBuffer = HornetQBuffers.dynamicBuffer(NettyConnection.BATCHING_BUFFER_SIZE); } } finally { - writeLock.set(false); + writeLock.release(); } } } @@ -177,73 +177,78 @@ public void write(HornetQBuffer buffer, final boolean flush, final boolean batched) { - while (!writeLock.compareAndSet(false, true)) - { - Thread.yield(); - } try { - if (batchBuffer == null && batchingEnabled && batched && !flush) - { - // Lazily create batch buffer + writeLock.acquire(); - batchBuffer = HornetQBuffers.dynamicBuffer(BATCHING_BUFFER_SIZE); - } - - if (batchBuffer != null) + try { - batchBuffer.writeBytes(buffer, 0, buffer.writerIndex()); - - if (batchBuffer.writerIndex() >= BATCHING_BUFFER_SIZE || !batched || flush) + if (batchBuffer == null && batchingEnabled && batched && !flush) { - // If the batch buffer is full or it's flush param or not batched then flush the buffer + // Lazily create batch buffer - buffer = batchBuffer; + batchBuffer = HornetQBuffers.dynamicBuffer(NettyConnection.BATCHING_BUFFER_SIZE); } - else - { - return; - } - if (!batched || flush) + if (batchBuffer != null) { - batchBuffer = null; - } - else - { - // Create a new buffer + batchBuffer.writeBytes(buffer, 0, buffer.writerIndex()); - batchBuffer = HornetQBuffers.dynamicBuffer(BATCHING_BUFFER_SIZE); + if (batchBuffer.writerIndex() >= NettyConnection.BATCHING_BUFFER_SIZE || !batched || flush) + { + // If the batch buffer is full or it's flush param or not batched then flush the buffer + + buffer = batchBuffer; + } + else + { + return; + } + + if (!batched || flush) + { + batchBuffer = null; + } + else + { + // Create a new buffer + + batchBuffer = HornetQBuffers.dynamicBuffer(NettyConnection.BATCHING_BUFFER_SIZE); + } } - } - ChannelFuture future = channel.write(buffer.channelBuffer()); + ChannelFuture future = channel.write(buffer.channelBuffer()); - if (flush) - { - while (true) + if (flush) { - try + while (true) { - boolean ok = future.await(10000); + try + { + boolean ok = future.await(10000); - if (!ok) + if (!ok) + { + NettyConnection.log.warn("Timed out waiting for packet to be flushed"); + } + + break; + } + catch (InterruptedException ignore) { - NettyConnection.log.warn("Timed out waiting for packet to be flushed"); } - - break; } - catch (InterruptedException ignore) - { - } } } + finally + { + writeLock.release(); + } } - finally + catch (InterruptedException e) { - writeLock.set(false); + Thread.currentThread().interrupt(); } } @@ -256,20 +261,20 @@ { return directDeliver; } - + public void addReadyListener(final ReadyListener listener) { readyListeners.add(listener); } - + public void removeReadyListener(final ReadyListener listener) { readyListeners.remove(listener); } - + public void fireReady(final boolean ready) { - for (ReadyListener listener: readyListeners) + for (ReadyListener listener : readyListeners) { listener.readyForWriting(ready); } Modified: branches/Branch_2_2_AS7/src/main/org/hornetq/core/server/cluster/Bridge.java =================================================================== --- branches/Branch_2_2_AS7/src/main/org/hornetq/core/server/cluster/Bridge.java 2011-11-07 04:32:51 UTC (rev 11665) +++ branches/Branch_2_2_AS7/src/main/org/hornetq/core/server/cluster/Bridge.java 2011-11-08 00:45:05 UTC (rev 11666) @@ -59,4 +59,6 @@ * Basically this is for cluster bridges being disconnected */ void disconnect(); + + boolean isConnected(); } Modified: branches/Branch_2_2_AS7/src/main/org/hornetq/core/server/cluster/ClusterConnection.java =================================================================== --- branches/Branch_2_2_AS7/src/main/org/hornetq/core/server/cluster/ClusterConnection.java 2011-11-07 04:32:51 UTC (rev 11665) +++ branches/Branch_2_2_AS7/src/main/org/hornetq/core/server/cluster/ClusterConnection.java 2011-11-08 00:45:05 UTC (rev 11666) @@ -65,4 +65,6 @@ void informTopology(); void announceBackup(); + + boolean isNodeActive(String id); } Modified: branches/Branch_2_2_AS7/src/main/org/hornetq/core/server/cluster/impl/BridgeImpl.java =================================================================== --- branches/Branch_2_2_AS7/src/main/org/hornetq/core/server/cluster/impl/BridgeImpl.java 2011-11-07 04:32:51 UTC (rev 11665) +++ branches/Branch_2_2_AS7/src/main/org/hornetq/core/server/cluster/impl/BridgeImpl.java 2011-11-08 00:45:05 UTC (rev 11666) @@ -25,9 +25,9 @@ import org.hornetq.api.core.Message; import org.hornetq.api.core.SimpleString; import org.hornetq.api.core.client.ClientProducer; +import org.hornetq.api.core.client.ClientSession.BindingQuery; import org.hornetq.api.core.client.SendAcknowledgementHandler; import org.hornetq.api.core.client.SessionFailureListener; -import org.hornetq.api.core.client.ClientSession.BindingQuery; import org.hornetq.api.core.management.NotificationType; import org.hornetq.core.client.impl.ClientSessionFactoryInternal; import org.hornetq.core.client.impl.ClientSessionInternal; @@ -46,7 +46,6 @@ import org.hornetq.core.server.management.Notification; import org.hornetq.core.server.management.NotificationService; import org.hornetq.spi.core.protocol.RemotingConnection; -import org.hornetq.utils.ConcurrentHashSet; import org.hornetq.utils.Future; import org.hornetq.utils.TypedProperties; import org.hornetq.utils.UUID; @@ -132,6 +131,8 @@ private NotificationService notificationService; + private boolean stopping = false; + // Static -------------------------------------------------------- // Constructors -------------------------------------------------- @@ -199,7 +200,7 @@ { this.notificationService = notificationService; } - + public synchronized void start() throws Exception { if (started) @@ -209,6 +210,8 @@ started = true; + stopping = false; + if (activated) { activate(); @@ -222,7 +225,7 @@ notificationService.sendNotification(notification); } } - + public String debug() { return toString(); @@ -305,20 +308,32 @@ }); } - /** The cluster manager needs to use the same executor to close the serverLocator, otherwise the stop will break. - * This method is intended to expose this executor to the ClusterManager */ + public boolean isConnected() + { + return session != null; + } + + /** The cluster manager needs to use the same executor to close the serverLocator, otherwise the stop will break. + * This method is intended to expose this executor to the ClusterManager */ public Executor getExecutor() { return executor; } - + public void stop() throws Exception { + if (stopping) + { + return; + } + + stopping = true; + if (log.isDebugEnabled()) { log.debug("Bridge " + this.name + " being stopped"); } - + if (futureScheduledReconnection != null) { futureScheduledReconnection.cancel(true); @@ -471,7 +486,10 @@ { if (log.isDebugEnabled()) { - log.debug("The transformer " + transformer + " made a copy of the message " + message + " as transformedMessage"); + log.debug("The transformer " + transformer + + " made a copy of the message " + + message + + " as transformedMessage"); } } return transformedMessage; @@ -544,12 +562,12 @@ // that this will throw a disconnect, we need to remove the message // from the acks so it will get resent, duplicate detection will cope // with any messages resent - + if (log.isTraceEnabled()) { log.trace("going to send message " + message); } - + try { producer.send(dest, message); @@ -580,7 +598,7 @@ { producer.close(); } - + csf.cleanup(); } catch (Throwable dontCare) @@ -681,7 +699,6 @@ return csf; } - /* Hook for creating session factory */ protected ClientSessionFactoryInternal createSessionFactory() throws Exception { @@ -702,6 +719,12 @@ if (csf == null || csf.isClosed()) { csf = createSessionFactory(); + if (csf == null) + { + // Retrying. This probably means the node is not available (for the cluster connection case) + scheduleRetryConnect(); + return; + } // Session is pre-acknowledge session = (ClientSessionInternal)csf.createSession(user, password, false, true, true, true, 1); } @@ -801,6 +824,12 @@ return; } + if (stopping) + { + log.info("Bridge is stopping, will not retry"); + return; + } + if (reconnectAttemptsInUse >= 0 && retryCount > reconnectAttempts) { log.warn("Bridge " + this.name + @@ -963,7 +992,10 @@ { public synchronized void run() { - connect(); + if (!stopping) + { + connect(); + } } } } Modified: branches/Branch_2_2_AS7/src/main/org/hornetq/core/server/cluster/impl/ClusterConnectionBridge.java =================================================================== --- branches/Branch_2_2_AS7/src/main/org/hornetq/core/server/cluster/impl/ClusterConnectionBridge.java 2011-11-07 04:32:51 UTC (rev 11665) +++ branches/Branch_2_2_AS7/src/main/org/hornetq/core/server/cluster/impl/ClusterConnectionBridge.java 2011-11-08 00:45:05 UTC (rev 11666) @@ -71,11 +71,9 @@ private final SimpleString idsHeaderName; private final String targetNodeID; - + private final long targetNodeEventUID; - private final TransportConfiguration connector; - private final ServerLocatorInternal discoveryLocator; public ClusterConnectionBridge(final ClusterConnection clusterConnection, @@ -138,7 +136,6 @@ this.managementAddress = managementAddress; this.managementNotificationAddress = managementNotificationAddress; this.flowRecord = flowRecord; - this.connector = connector; // we need to disable DLQ check on the clustered bridges queue.setInternalQueue(true); @@ -152,7 +149,15 @@ protected ClientSessionFactoryInternal createSessionFactory() throws Exception { - ClientSessionFactoryInternal factory = super.createSessionFactory(); + ClientSessionFactoryInternal factory = (ClientSessionFactoryInternal)serverLocator.createSessionFactory(targetNodeID); + + if (factory == null) + { + log.warn("NodeID=" + targetNodeID + + " is not available on the topology. Retrying the connection to that node now"); + return null; + } + factory.setReconnectAttempts(0); factory.getConnection().addFailureListener(new FailureListener() { @@ -177,7 +182,7 @@ // nodes could have same queue ids // Note we must copy since same message may get routed to other nodes which require different headers ServerMessage messageCopy = message.copy(); - + if (log.isTraceEnabled()) { log.trace("Clustered bridge copied message " + message + " as " + messageCopy + " before delivery"); @@ -188,7 +193,7 @@ Set<SimpleString> propNames = new HashSet<SimpleString>(messageCopy.getPropertyNames()); byte[] queueIds = message.getBytesProperty(idsHeaderName); - + if (queueIds == null) { // Sanity check only @@ -211,7 +216,7 @@ messageCopy.putBytesProperty(MessageImpl.HDR_ROUTE_TO_IDS, queueIds); messageCopy = super.beforeForward(messageCopy); - + return messageCopy; } @@ -339,7 +344,7 @@ if (permanently) { log.debug("cluster node for bridge " + this.getName() + " is permanently down"); - discoveryLocator.notifyNodeDown(targetNodeEventUID+1, targetNodeID); + discoveryLocator.notifyNodeDown(System.currentTimeMillis(), targetNodeID); } } Modified: branches/Branch_2_2_AS7/src/main/org/hornetq/core/server/cluster/impl/ClusterConnectionImpl.java =================================================================== --- branches/Branch_2_2_AS7/src/main/org/hornetq/core/server/cluster/impl/ClusterConnectionImpl.java 2011-11-07 04:32:51 UTC (rev 11665) +++ branches/Branch_2_2_AS7/src/main/org/hornetq/core/server/cluster/impl/ClusterConnectionImpl.java 2011-11-08 00:45:05 UTC (rev 11666) @@ -143,20 +143,23 @@ private final ClusterManagerInternal manager; - + private final int minLargeMessageSize; + + // Stuff that used to be on the ClusterManager - private final Topology topology = new Topology(this); private volatile ServerLocatorInternal backupServerLocator; + private boolean stopping = false; public ClusterConnectionImpl(final ClusterManagerInternal manager, final TransportConfiguration[] tcConfigs, final TransportConfiguration connector, final SimpleString name, final SimpleString address, + final int minLargeMessageSize, final long clientFailureCheckPeriod, final long connectionTTL, final long retryInterval, @@ -214,7 +217,7 @@ this.executorFactory = executorFactory; this.executor = executorFactory.getExecutor(); - + this.topology.setExecutor(executor); this.server = server; @@ -238,9 +241,19 @@ this.manager = manager; this.callTimeout = callTimeout; + + this.minLargeMessageSize = minLargeMessageSize; clusterConnector = new StaticClusterConnector(tcConfigs); + backupServerLocator = clusterConnector.createServerLocator(false); + + if (backupServerLocator != null) + { + backupServerLocator.setReconnectAttempts(-1); + backupServerLocator.setInitialConnectAttempts(-1); + } + if (tcConfigs != null && tcConfigs.length > 0) { // a cluster connection will connect to other nodes only if they are directly connected @@ -258,6 +271,7 @@ final TransportConfiguration connector, final SimpleString name, final SimpleString address, + final int minLargeMessageSize, final long clientFailureCheckPeriod, final long connectionTTL, final long retryInterval, @@ -303,6 +317,8 @@ this.retryIntervalMultiplier = retryIntervalMultiplier; this.maxRetryInterval = maxRetryInterval; + + this.minLargeMessageSize = minLargeMessageSize; this.reconnectAttempts = reconnectAttempts; @@ -317,7 +333,7 @@ this.executorFactory = executorFactory; this.executor = executorFactory.getExecutor(); - + this.topology.setExecutor(executor); this.server = server; @@ -340,10 +356,18 @@ clusterConnector = new DiscoveryClusterConnector(dg); + backupServerLocator = clusterConnector.createServerLocator(true); + + if (backupServerLocator != null) + { + backupServerLocator.setReconnectAttempts(-1); + backupServerLocator.setInitialConnectAttempts(-1); + } + this.manager = manager; } - public void start() throws Exception + public void start() throws Exception { synchronized (this) { @@ -351,10 +375,10 @@ { return; } - - + + stopping = false; started = true; - + if (!backup) { activate(); @@ -362,7 +386,7 @@ } } - + public void flushExecutor() { Future future = new Future(); @@ -379,7 +403,7 @@ { return; } - + stopping = true; if (log.isDebugEnabled()) { log.debug(this + "::stopping ClusterConnection"); @@ -419,9 +443,7 @@ props); managementService.sendNotification(notification); } - - executor.execute(new Runnable() { public void run() @@ -447,15 +469,8 @@ started = false; } - public void announceBackup() { - this.backupServerLocator = clusterConnector.createServerLocator(false); - - backupServerLocator.setReconnectAttempts(-1); - backupServerLocator.setInitialConnectAttempts(-1); - - executor.execute(new Runnable() { public void run() @@ -491,7 +506,7 @@ { return topology.getMember(manager.getNodeId()); } - + public void addClusterTopologyListener(final ClusterTopologyListener listener, final boolean clusterConnection) { topology.addClusterTopologyListener(listener); @@ -509,7 +524,7 @@ { return topology; } - + public void nodeAnnounced(final long uniqueEventID, final String nodeID, final Pair<TransportConfiguration, TransportConfiguration> connectorPair, @@ -571,6 +586,16 @@ return server; } + public boolean isNodeActive(String nodeId) + { + MessageFlowRecord rec = records.get(nodeId); + if (rec == null) + { + return false; + } + return rec.getBridge().isConnected(); + } + public Map<String, String> getNodes() { synchronized (records) @@ -600,7 +625,7 @@ } backup = false; - + topology.updateAsLive(manager.getNodeId(), new TopologyMember(connector, null)); if (backupServerLocator != null) @@ -617,8 +642,6 @@ backupServerLocator = null; } - - serverLocator = clusterConnector.createServerLocator(true); if (serverLocator != null) @@ -686,6 +709,10 @@ public void nodeDown(final long eventUID, final String nodeID) { + if (stopping) + { + return; + } if (log.isDebugEnabled()) { log.debug(this + " receiving nodeDown for nodeID=" + nodeID, new Exception("trace")); @@ -721,6 +748,10 @@ final Pair<TransportConfiguration, TransportConfiguration> connectorPair, final boolean last) { + if (stopping) + { + return; + } if (log.isDebugEnabled()) { String ClusterTestBase = "receiving nodeUP for nodeID="; @@ -821,13 +852,13 @@ } } } - + public synchronized void informTopology() { String nodeID = server.getNodeID().toString(); - + TopologyMember localMember; - + if (backup) { localMember = new TopologyMember(null, connector); @@ -840,7 +871,6 @@ topology.updateAsLive(nodeID, localMember); } - private void createNewRecord(final long eventUID, final String targetNodeID, final TransportConfiguration connector, @@ -848,22 +878,22 @@ final Queue queue, final boolean start) throws Exception { - final ServerLocatorInternal targetLocator = new ServerLocatorImpl(topology, false, connector); - + final ServerLocatorInternal targetLocator = new ServerLocatorImpl(topology, true, connector); + String nodeId; - + synchronized (this) { if (!started) { return; } - + if (serverLocator == null) { return; } - + nodeId = serverLocator.getNodeID(); } @@ -881,6 +911,7 @@ targetLocator.setRetryInterval(retryInterval); targetLocator.setMaxRetryInterval(maxRetryInterval); targetLocator.setRetryIntervalMultiplier(retryIntervalMultiplier); + targetLocator.setMinLargeMessageSize(minLargeMessageSize); targetLocator.setAfterConnectionInternalListener(this); @@ -1125,9 +1156,9 @@ public synchronized void onMessage(final ClientMessage message) { - if (isTrace) + if (log.isDebugEnabled()) { - log.trace("Flow record on " + clusterConnector + " Receiving message " + message); + log.debug("ClusterCommunication::Flow record on " + clusterConnector + " Receiving message " + message); } try { @@ -1504,8 +1535,9 @@ @Override public String toString() { - return "ClusterConnectionImpl@" + System.identityHashCode(this) + - "[nodeUUID=" + nodeUUID + + return "ClusterConnectionImpl@" + System.identityHashCode(this) + + "[nodeUUID=" + + nodeUUID + ", connector=" + connector + ", address=" + @@ -1555,7 +1587,7 @@ { log.debug(ClusterConnectionImpl.this + "Creating a serverLocator for " + Arrays.toString(tcConfigs)); } - ServerLocatorImpl locator = new ServerLocatorImpl(includeTopology?topology:null, true, tcConfigs); + ServerLocatorImpl locator = new ServerLocatorImpl(includeTopology ? topology : null, true, tcConfigs); locator.setClusterConnection(true); return locator; } @@ -1587,7 +1619,7 @@ public ServerLocatorInternal createServerLocator(boolean includeTopology) { - ServerLocatorImpl locator = new ServerLocatorImpl(includeTopology?topology:null, true, dg); + ServerLocatorImpl locator = new ServerLocatorImpl(includeTopology ? topology : null, true, dg); return locator; } Modified: branches/Branch_2_2_AS7/src/main/org/hornetq/core/server/cluster/impl/ClusterManagerImpl.java =================================================================== --- branches/Branch_2_2_AS7/src/main/org/hornetq/core/server/cluster/impl/ClusterManagerImpl.java 2011-11-07 04:32:51 UTC (rev 11665) +++ branches/Branch_2_2_AS7/src/main/org/hornetq/core/server/cluster/impl/ClusterManagerImpl.java 2011-11-08 00:45:05 UTC (rev 11666) @@ -470,6 +470,7 @@ serverLocator.setClientFailureCheckPeriod(config.getClientFailureCheckPeriod()); serverLocator.setBlockOnDurableSend(!config.isUseDuplicateDetection()); serverLocator.setBlockOnNonDurableSend(!config.isUseDuplicateDetection()); + serverLocator.setMinLargeMessageSize(config.getMinLargeMessageSize()); if (!config.isUseDuplicateDetection()) { log.debug("Bridge " + config.getName() + @@ -622,6 +623,7 @@ connector, new SimpleString(config.getName()), new SimpleString(config.getAddress()), + config.getMinLargeMessageSize(), config.getClientFailureCheckPeriod(), config.getConnectionTTL(), config.getRetryInterval(), @@ -659,6 +661,7 @@ connector, new SimpleString(config.getName()), new SimpleString(config.getAddress()), + config.getMinLargeMessageSize(), config.getClientFailureCheckPeriod(), config.getConnectionTTL(), config.getRetryInterval(), Modified: branches/Branch_2_2_AS7/src/main/org/hornetq/core/server/cluster/impl/RemoteQueueBindingImpl.java =================================================================== --- branches/Branch_2_2_AS7/src/main/org/hornetq/core/server/cluster/impl/RemoteQueueBindingImpl.java 2011-11-07 04:32:51 UTC (rev 11665) +++ branches/Branch_2_2_AS7/src/main/org/hornetq/core/server/cluster/impl/RemoteQueueBindingImpl.java 2011-11-08 00:45:05 UTC (rev 11666) @@ -311,6 +311,11 @@ buff.putLong(remoteQueueID); message.putBytesProperty(idsHeaderName, ids); + + if (log.isTraceEnabled()) + { + log.trace("Adding remoteQueue ID = " + remoteQueueID + " into message=" + message + " store-forward-queue=" + storeAndForwardQueue); + } } } Modified: branches/Branch_2_2_AS7/src/main/org/hornetq/core/server/impl/DivertImpl.java =================================================================== --- branches/Branch_2_2_AS7/src/main/org/hornetq/core/server/impl/DivertImpl.java 2011-11-07 04:32:51 UTC (rev 11665) +++ branches/Branch_2_2_AS7/src/main/org/hornetq/core/server/impl/DivertImpl.java 2011-11-08 00:45:05 UTC (rev 11666) @@ -84,6 +84,11 @@ // properly on ack, since the original address will be overwritten // TODO we can optimise this so it doesn't copy if it's not routed anywhere else + + if (log.isTraceEnabled()) + { + log.trace("Diverting message " + message + " into " + this); + } long id = storageManager.generateUniqueID(); Modified: branches/Branch_2_2_AS7/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java =================================================================== --- branches/Branch_2_2_AS7/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java 2011-11-07 04:32:51 UTC (rev 11665) +++ branches/Branch_2_2_AS7/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java 2011-11-08 00:45:05 UTC (rev 11666) @@ -317,6 +317,7 @@ public synchronized void start() throws Exception { + log.debug("Starting server " + this); OperationContextImpl.clearContext(); try Modified: branches/Branch_2_2_AS7/src/main/org/hornetq/core/server/impl/QueueImpl.java =================================================================== --- branches/Branch_2_2_AS7/src/main/org/hornetq/core/server/impl/QueueImpl.java 2011-11-07 04:32:51 UTC (rev 11665) +++ branches/Branch_2_2_AS7/src/main/org/hornetq/core/server/impl/QueueImpl.java 2011-11-08 00:45:05 UTC (rev 11666) @@ -461,14 +461,14 @@ { if (isTrace) { - log.trace("Force delivery scheduling depage"); + log.trace("Force delivery scheduling depage"); } scheduleDepage(); } if (isTrace) { - log.trace("Force delivery deliverying async"); + log.trace("Force delivery deliverying async"); } deliverAsync(); @@ -2255,7 +2255,7 @@ } catch (Exception e) { - QueueImpl.log.warn("Unable to decrement reference counting", e); + QueueImpl.log.warn("Unable to decrement reference counting", e); } } Modified: branches/Branch_2_2_AS7/src/main/org/hornetq/jms/client/HornetQConnectionFactory.java =================================================================== --- branches/Branch_2_2_AS7/src/main/org/hornetq/jms/client/HornetQConnectionFactory.java 2011-11-07 04:32:51 UTC (rev 11665) +++ branches/Branch_2_2_AS7/src/main/org/hornetq/jms/client/HornetQConnectionFactory.java 2011-11-08 00:45:05 UTC (rev 11666) @@ -700,6 +700,24 @@ // Private -------------------------------------------------------------------------------------- + /* (non-Javadoc) + * @see java.lang.Object#toString() + */ + @Override + public String toString() + { + return "HornetQConnectionFactory [serverLocator=" + serverLocator + + ", clientID=" + + clientID + + ", dupsOKBatchSize=" + + dupsOKBatchSize + + ", transactionBatchSize=" + + transactionBatchSize + + ", readOnly=" + + readOnly + + "]"; + } + private void checkWrite() { if (readOnly) Modified: branches/Branch_2_2_AS7/src/main/org/hornetq/jms/server/recovery/HornetQXAResourceWrapper.java =================================================================== --- branches/Branch_2_2_AS7/src/main/org/hornetq/jms/server/recovery/HornetQXAResourceWrapper.java 2011-11-07 04:32:51 UTC (rev 11665) +++ branches/Branch_2_2_AS7/src/main/org/hornetq/jms/server/recovery/HornetQXAResourceWrapper.java 2011-11-08 00:45:05 UTC (rev 11666) @@ -13,17 +13,15 @@ package org.hornetq.jms.server.recovery; -import java.util.Map; +import java.util.Arrays; import javax.transaction.xa.XAException; import javax.transaction.xa.XAResource; import javax.transaction.xa.Xid; import org.hornetq.api.core.HornetQException; -import org.hornetq.api.core.TransportConfiguration; import org.hornetq.api.core.client.ClientSession; import org.hornetq.api.core.client.ClientSessionFactory; -import org.hornetq.api.core.client.HornetQClient; import org.hornetq.api.core.client.ServerLocator; import org.hornetq.api.core.client.SessionFailureListener; import org.hornetq.core.logging.Logger; @@ -53,19 +51,25 @@ private static final Object lock = new Object(); private ServerLocator serverLocator; - + private ClientSessionFactory csf; private XAResource delegate; private XARecoveryConfig[] xaRecoveryConfigs; - //private TransportConfiguration currentConnection; + // private TransportConfiguration currentConnection; public HornetQXAResourceWrapper(XARecoveryConfig... xaRecoveryConfigs) { + this.xaRecoveryConfigs = xaRecoveryConfigs; - this.xaRecoveryConfigs = xaRecoveryConfigs; + if (log.isDebugEnabled()) + { + log.debug("Recovery configured with " + Arrays.toString(xaRecoveryConfigs) + + ", instance=" + + System.identityHashCode(this)); + } } public Xid[] recover(final int flag) throws XAException @@ -74,10 +78,18 @@ HornetQXAResourceWrapper.log.debug("Recover " + xaResource); try { - return xaResource.recover(flag); + Xid[] xids = xaResource.recover(flag); + + if (log.isDebugEnabled() && xids != null && xids.length > 0) + { + log.debug("Recovering these following IDs " + Arrays.toString(xids) + " at " + this); + } + + return xids; } catch (XAException e) { + log.warn(e.getMessage(), e); throw check(e); } } @@ -214,7 +226,8 @@ public void connectionFailed(final HornetQException me, boolean failedOver) { - HornetQXAResourceWrapper.log.warn("Notified of connection failure in xa recovery connectionFactory for provider " + csf + " will attempt reconnect on next pass", + HornetQXAResourceWrapper.log.warn("Notified of connection failure in xa recovery connectionFactory for provider " + csf + + " will attempt reconnect on next pass", me); close(); } @@ -244,9 +257,9 @@ if (result == null) { - //we should always throw a retry for certain methods comit etc, if not the tx is marked as a heuristic and - //all chaos is let loose - if(retry) + // we should always throw a retry for certain methods comit etc, if not the tx is marked as a heuristic and + // all chaos is let loose + if (retry) { XAException xae = new XAException("Connection unavailable for xa recovery"); xae.errorCode = XAException.XA_RETRY; @@ -294,6 +307,10 @@ for (XARecoveryConfig xaRecoveryConfig : xaRecoveryConfigs) { + if (log.isDebugEnabled()) + { + log.debug("Trying to connect recovery on " + xaRecoveryConfig + " of " + Arrays.toString(xaRecoveryConfigs)); + } ClientSession cs = null; @@ -308,7 +325,13 @@ } else { - cs = csf.createSession(xaRecoveryConfig.getUsername(), xaRecoveryConfig.getPassword(), true, false, false, false, 1); + cs = csf.createSession(xaRecoveryConfig.getUsername(), + xaRecoveryConfig.getPassword(), + true, + false, + false, + false, + 1); } } catch (HornetQException e) @@ -323,10 +346,29 @@ } return delegate; - } + } + log.warn("Can't connect to any hornetq server on recovery " + Arrays.toString(xaRecoveryConfigs)); throw new HornetQException(HornetQException.NOT_CONNECTED); } + /* (non-Javadoc) + * @see java.lang.Object#toString() + */ + @Override + public String toString() + { + return "HornetQXAResourceWrapper [serverLocator=" + serverLocator + + ", csf=" + + csf + + ", delegate=" + + delegate + + ", xaRecoveryConfigs=" + + Arrays.toString(xaRecoveryConfigs) + + ", instance=" + + System.identityHashCode(this) + + "]"; + } + /** * Close the connection */ @@ -366,6 +408,8 @@ */ protected XAException check(final XAException e) throws XAException { + log.warn(e.getMessage(), e); + if (e.errorCode == XAException.XA_RETRY) { close(); Modified: branches/Branch_2_2_AS7/src/main/org/hornetq/jms/server/recovery/XARecoveryConfig.java =================================================================== --- branches/Branch_2_2_AS7/src/main/org/hornetq/jms/server/recovery/XARecoveryConfig.java 2011-11-07 04:32:51 UTC (rev 11665) +++ branches/Branch_2_2_AS7/src/main/org/hornetq/jms/server/recovery/XARecoveryConfig.java 2011-11-08 00:45:05 UTC (rev 11666) @@ -65,8 +65,24 @@ return true; } + + + /* (non-Javadoc) + * @see java.lang.Object#toString() + */ @Override + public String toString() + { + return "XARecoveryConfig [hornetQConnectionFactory=" + hornetQConnectionFactory + + ", username=" + + username + + ", password=" + + password + + "]"; + } + + @Override public int hashCode() { int result = hornetQConnectionFactory != null ? hornetQConnectionFactory.hashCode() : 0; Modified: branches/Branch_2_2_AS7/src/main/org/hornetq/ra/HornetQRAManagedConnectionFactory.java =================================================================== --- branches/Branch_2_2_AS7/src/main/org/hornetq/ra/HornetQRAManagedConnectionFactory.java 2011-11-07 04:32:51 UTC (rev 11665) +++ branches/Branch_2_2_AS7/src/main/org/hornetq/ra/HornetQRAManagedConnectionFactory.java 2011-11-08 00:45:05 UTC (rev 11666) @@ -138,6 +138,11 @@ cm); } + if (connectionFactory == null) + { + connectionFactory = ra.createHornetQConnectionFactory(mcfProperties); + resourceRecovery = ra.getRecoveryManager().register(connectionFactory, null, null); + } return cf; } Modified: branches/Branch_2_2_AS7/src/main/org/hornetq/ra/HornetQResourceAdapter.java =================================================================== --- branches/Branch_2_2_AS7/src/main/org/hornetq/ra/HornetQResourceAdapter.java 2011-11-07 04:32:51 UTC (rev 11665) +++ branches/Branch_2_2_AS7/src/main/org/hornetq/ra/HornetQResourceAdapter.java 2011-11-08 00:45:05 UTC (rev 11666) @@ -218,6 +218,18 @@ this.ctx = ctx; + if (!configured.getAndSet(true)) + { + try + { + setup(); + } + catch (HornetQException e) + { + throw new ResourceAdapterInternalException("Unable to create activation", e); + } + } + HornetQResourceAdapter.log.info("HornetQ resource adaptor started"); } Modified: branches/Branch_2_2_AS7/src/main/org/hornetq/ra/recovery/RecoveryManager.java =================================================================== --- branches/Branch_2_2_AS7/src/main/org/hornetq/ra/recovery/RecoveryManager.java 2011-11-07 04:32:51 UTC (rev 11665) +++ branches/Branch_2_2_AS7/src/main/org/hornetq/ra/recovery/RecoveryManager.java 2011-11-08 00:45:05 UTC (rev 11666) @@ -54,6 +54,8 @@ public HornetQResourceRecovery register(HornetQConnectionFactory factory, String userName, String password) { + log.debug("registering recovery for factory : " + factory); + if(!isRegistered(factory) && registry != null) { XARecoveryConfig xaRecoveryConfig = new XARecoveryConfig(factory, userName, password); Modified: branches/Branch_2_2_AS7/tests/config/ConfigurationTest-full-config.xml =================================================================== --- branches/Branch_2_2_AS7/tests/config/ConfigurationTest-full-config.xml 2011-11-07 04:32:51 UTC (rev 11665) +++ branches/Branch_2_2_AS7/tests/config/ConfigurationTest-full-config.xml 2011-11-08 00:45:05 UTC (rev 11666) @@ -147,6 +147,7 @@ <forwarding-address>bridge-forwarding-address1</forwarding-address> <filter string="sku > 1"/> <transformer-class-name>org.foo.BridgeTransformer</transformer-class-name> + <min-large-message-size>4</min-large-message-size> <retry-interval>3</retry-interval> <retry-interval-multiplier>0.2</retry-interval-multiplier> <reconnect-attempts>2</reconnect-attempts> @@ -166,6 +167,7 @@ <cluster-connection name="cluster-connection1"> <address>queues1</address> <connector-ref>connector1</connector-ref> + <min-large-message-size>321</min-large-message-size> <call-timeout>123</call-timeout> <retry-interval>3</retry-interval> <use-duplicate-detection>true</use-duplicate-detection> Modified: branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/client/IncompatibleVersionTest.java =================================================================== --- branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/client/IncompatibleVersionTest.java 2011-11-07 04:32:51 UTC (rev 11665) +++ branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/client/IncompatibleVersionTest.java 2011-11-08 00:45:05 UTC (rev 11666) @@ -14,14 +14,10 @@ package org.hornetq.tests.integration.client; import static org.hornetq.tests.util.RandomUtil.randomString; -import org.hornetq.tests.util.SpawnedVMSupport; import java.io.FileOutputStream; import java.io.InputStream; import java.util.Properties; -import java.util.concurrent.ExecutorService; -import java.util.concurrent.Executors; -import java.util.concurrent.ScheduledExecutorService; import org.hornetq.api.core.HornetQException; import org.hornetq.api.core.TransportConfiguration; @@ -37,12 +33,12 @@ import org.hornetq.core.protocol.core.Packet; import org.hornetq.core.protocol.core.impl.wireformat.CreateSessionMessage; import org.hornetq.core.protocol.core.impl.wireformat.CreateSessionResponseMessage; -import org.hornetq.core.remoting.impl.invm.InVMConnectorFactory; import org.hornetq.core.remoting.server.impl.RemotingServiceImpl; import org.hornetq.core.server.HornetQServer; import org.hornetq.core.server.HornetQServers; import org.hornetq.core.version.impl.VersionImpl; import org.hornetq.tests.util.ServiceTestBase; +import org.hornetq.tests.util.SpawnedVMSupport; import org.hornetq.utils.VersionLoader; /** @@ -55,15 +51,19 @@ public class IncompatibleVersionTest extends ServiceTestBase { private static final Logger log = Logger.getLogger(IncompatibleVersionTest.class); + // Constants ----------------------------------------------------- // Attributes ---------------------------------------------------- private HornetQServer server; - private CoreRemotingConnection connection; private ServerLocator locator; + private ClientSessionFactory csf; + + private CoreRemotingConnection connection; + // Static -------------------------------------------------------- // Constructors -------------------------------------------------- @@ -73,12 +73,13 @@ @Override protected void setUp() throws Exception { + super.setUp(); server = createServer(false, false); server.getConfiguration().setConnectionTTLOverride(500); server.start(); locator = HornetQClient.createServerLocatorWithoutHA(new TransportConfiguration(ServiceTestBase.INVM_CONNECTOR_FACTORY)); - ClientSessionFactory csf = locator.createSessionFactory(); + csf = locator.createSessionFactory(); connection = csf.getConnection(); } @@ -86,11 +87,12 @@ @Override protected void tearDown() throws Exception { - connection.destroy(); + csf.close(); locator.close(); server.stop(); + super.tearDown(); } public void testCompatibleClientVersion() throws Exception @@ -103,20 +105,36 @@ doTestClientVersionCompatibility(false); } - public void testCompatibleClientVersionWithRealConnection() throws Exception + public void testCompatibleClientVersionWithRealConnection1() throws Exception { - assertTrue(doTestClientVersionCompatibilityWithRealConnection("1-3,5,7-10",1)); - assertTrue(doTestClientVersionCompatibilityWithRealConnection("1-3,5,7-10",5)); - assertTrue(doTestClientVersionCompatibilityWithRealConnection("1-3,5,7-10",10)); + assertTrue(doTestClientVersionCompatibilityWithRealConnection("1-3,5,7-10", 1)); } - public void testIncompatibleClientVersionWithRealConnection() throws Exception + public void testCompatibleClientVersionWithRealConnection2() throws Exception { - assertFalse(doTestClientVersionCompatibilityWithRealConnection("1-3,5,7-10",0)); - assertFalse(doTestClientVersionCompatibilityWithRealConnection("1-3,5,7-10",4)); - assertFalse(doTestClientVersionCompatibilityWithRealConnection("1-3,5,7-10",100)); + assertTrue(doTestClientVersionCompatibilityWithRealConnection("1-3,5,7-10", 5)); } - + + public void testCompatibleClientVersionWithRealConnection3() throws Exception + { + assertTrue(doTestClientVersionCompatibilityWithRealConnection("1-3,5,7-10", 10)); + } + + public void testIncompatibleClientVersionWithRealConnection1() throws Exception + { + assertFalse(doTestClientVersionCompatibilityWithRealConnection("1-3,5,7-10", 0)); + } + + public void testIncompatibleClientVersionWithRealConnection2() throws Exception + { + assertFalse(doTestClientVersionCompatibilityWithRealConnection("1-3,5,7-10", 4)); + } + + public void testIncompatibleClientVersionWithRealConnection3() throws Exception + { + assertFalse(doTestClientVersionCompatibilityWithRealConnection("1-3,5,7-10", 100)); + } + private void doTestClientVersionCompatibility(boolean compatible) throws Exception { Channel channel1 = connection.getChannel(1, -1); @@ -175,61 +193,69 @@ { String propFileName = "compatibility-test-hornetq-version.properties"; String serverStartedString = "IncompatibleVersionTest---server---started"; - + Properties prop = new Properties(); InputStream in = VersionImpl.class.getClassLoader().getResourceAsStream("hornetq-version.properties"); prop.load(in); prop.setProperty("hornetq.version.compatibleVersionList", verList); prop.setProperty("hornetq.version.incrementingVersion", Integer.toString(ver)); prop.store(new FileOutputStream("tests/tmpfiles/" + propFileName), null); - + Process server = null; boolean result = false; try { server = SpawnedVMSupport.spawnVM("org.hornetq.tests.integration.client.IncompatibleVersionTest", - new String[]{"-D" + VersionLoader.VERSION_PROP_FILE_KEY + "=" + propFileName}, + new String[] { "-D" + VersionLoader.VERSION_PROP_FILE_KEY + + "=" + + propFileName }, "server", serverStartedString); Thread.sleep(2000); - + Process client = SpawnedVMSupport.spawnVM("org.hornetq.tests.integration.client.IncompatibleVersionTest", - new String[]{"-D" + VersionLoader.VERSION_PROP_FILE_KEY + "=" + propFileName}, + new String[] { "-D" + VersionLoader.VERSION_PROP_FILE_KEY + + "=" + + propFileName }, "client"); - - if(client.waitFor() == 0) + + if (client.waitFor() == 0) { result = true; } } finally { - if(server != null) + if (server != null) { try { server.destroy(); } - catch(Throwable t) {/* ignore */} + catch (Throwable t) + {/* ignore */ + } } } - + return result; } - + private static class ServerStarter { public void perform(String startedString) throws Exception { Configuration conf = new ConfigurationImpl(); conf.setSecurityEnabled(false); - conf.getAcceptorConfigurations().add(new TransportConfiguration("org.hornetq.core.remoting.impl.netty.NettyAcceptorFactory")); + conf.getAcceptorConfigurations() + .add(new TransportConfiguration("org.hornetq.core.remoting.impl.netty.NettyAcceptorFactory")); HornetQServer server = HornetQServers.newHornetQServer(conf, false); server.start(); - + log.info("### server: " + startedString); } } + private static class ClientStarter { public void perform() throws Exception @@ -243,15 +269,15 @@ locator.close(); } } - + public static void main(String[] args) throws Exception { - if(args[0].equals("server")) + if (args[0].equals("server")) { ServerStarter ss = new ServerStarter(); ss.perform(args[1]); } - else if(args[0].equals("client")) + else if (args[0].equals("client")) { ClientStarter cs = new ClientStarter(); cs.perform(); @@ -261,7 +287,7 @@ throw new Exception("args[0] must be \"server\" or \"client\""); } } - + // Package protected --------------------------------------------- // Protected ----------------------------------------------------- Modified: branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/client/InterruptedLargeMessageTest.java =================================================================== --- branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/client/InterruptedLargeMessageTest.java 2011-11-07 04:32:51 UTC (rev 11665) +++ branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/client/InterruptedLargeMessageTest.java 2011-11-08 00:45:05 UTC (rev 11666) @@ -13,36 +13,30 @@ package org.hornetq.tests.integration.client; -import java.io.ByteArrayOutputStream; import java.util.HashMap; import javax.transaction.xa.XAResource; import javax.transaction.xa.Xid; -import junit.framework.Assert; - import org.hornetq.api.core.HornetQException; import org.hornetq.api.core.Interceptor; import org.hornetq.api.core.Message; import org.hornetq.api.core.SimpleString; -import org.hornetq.api.core.client.*; -import org.hornetq.core.client.impl.ClientConsumerInternal; -import org.hornetq.core.config.Configuration; +import org.hornetq.api.core.client.ClientConsumer; +import org.hornetq.api.core.client.ClientMessage; +import org.hornetq.api.core.client.ClientProducer; +import org.hornetq.api.core.client.ClientSession; +import org.hornetq.api.core.client.ClientSessionFactory; +import org.hornetq.api.core.client.HornetQClient; +import org.hornetq.api.core.client.ServerLocator; import org.hornetq.core.logging.Logger; -import org.hornetq.core.persistence.impl.journal.JournalStorageManager; -import org.hornetq.core.persistence.impl.journal.LargeServerMessageImpl; -import org.hornetq.core.persistence.impl.journal.OperationContextImpl; import org.hornetq.core.protocol.core.Packet; import org.hornetq.core.protocol.core.impl.wireformat.SessionSendContinuationMessage; -import org.hornetq.core.server.HornetQServer; -import org.hornetq.core.server.Queue; import org.hornetq.core.server.ServerSession; import org.hornetq.core.server.impl.ServerSessionImpl; import org.hornetq.core.settings.impl.AddressSettings; import org.hornetq.spi.core.protocol.RemotingConnection; import org.hornetq.tests.integration.largemessage.LargeMessageTestBase; -import org.hornetq.tests.util.RandomUtil; -import org.hornetq.tests.util.UnitTestCase; /** * A LargeMessageTest @@ -332,9 +326,9 @@ server.stop(false); server.start(); - server.stop(); - validateNoFilesOnLargeDir(); + + server.stop(); } finally { Modified: branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/client/JournalCrashTest.java =================================================================== --- branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/client/JournalCrashTest.java 2011-11-07 04:32:51 UTC (rev 11665) +++ branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/client/JournalCrashTest.java 2011-11-08 00:45:05 UTC (rev 11666) @@ -127,20 +127,15 @@ { try { - int start = 4; - int end = 8; + int start = Integer.parseInt(arg[1]); - if (arg.length > 0) - { - start = Integer.parseInt(arg[0]); - } + int end = Integer.parseInt(arg[2]); + + long timeStart = Long.parseLong(arg[0]); - if (arg.length > 1) - { - end = Integer.parseInt(arg[1]); - } - JournalCrashTest restart = new JournalCrashTest(); + + restart.setTimeStart(timeStart); restart.startServer(); @@ -196,10 +191,10 @@ public void testRestartJournal() throws Throwable { - runExternalProcess(0, JournalCrashTest.FIRST_RUN); - runExternalProcess(JournalCrashTest.FIRST_RUN, JournalCrashTest.SECOND_RUN); - runExternalProcess(JournalCrashTest.SECOND_RUN, JournalCrashTest.THIRD_RUN); - runExternalProcess(JournalCrashTest.THIRD_RUN, JournalCrashTest.FOURTH_RUN); + runExternalProcess(getTimeStart(), 0, JournalCrashTest.FIRST_RUN); + runExternalProcess(getTimeStart(), JournalCrashTest.FIRST_RUN, JournalCrashTest.SECOND_RUN); + runExternalProcess(getTimeStart(), JournalCrashTest.SECOND_RUN, JournalCrashTest.THIRD_RUN); + runExternalProcess(getTimeStart(), JournalCrashTest.THIRD_RUN, JournalCrashTest.FOURTH_RUN); printJournal(); @@ -241,7 +236,7 @@ * @throws Exception * @throws InterruptedException */ - private void runExternalProcess(final int start, final int end) throws Exception, InterruptedException + private void runExternalProcess(final long timeStart, final int start, final int end) throws Exception, InterruptedException { System.err.println("running external process..."); Process process = SpawnedVMSupport.spawnVM(this.getClass().getCanonicalName(), @@ -249,6 +244,7 @@ new String[] {}, true, true, + Long.toString(timeStart), Integer.toString(start), Integer.toString(end)); Modified: branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/client/LargeMessageCompressTest.java =================================================================== --- branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/client/LargeMessageCompressTest.java 2011-11-07 04:32:51 UTC (rev 11665) +++ branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/client/LargeMessageCompressTest.java 2011-11-08 00:45:05 UTC (rev 11666) @@ -159,8 +159,10 @@ msg1.setOutputStream(output); - msg1.waitOutputStreamCompletion(0); + msg1.waitOutputStreamCompletion(0); + output.close(); + msg1.acknowledge(); session.commit(); @@ -176,7 +178,7 @@ byte b = (byte)input.read(); assertEquals("position = " + i, getSamplebyte(i), b); } - + input.close(); testFile.delete(); validateNoFilesOnLargeDir(); } @@ -239,6 +241,8 @@ msg1.saveToOutputStream(output); msg1.acknowledge(); + + output.close(); session.commit(); @@ -253,6 +257,7 @@ byte b = (byte)input.read(); assertEquals("position = " + i, getSamplebyte(i), b); } + input.close(); testFile.delete(); validateNoFilesOnLargeDir(); @@ -358,6 +363,7 @@ byte b = (byte)input.read(); assertEquals("position = " + i, msgs[i], b); } + input.close(); testFile.delete(); validateNoFilesOnLargeDir(); Modified: branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/client/LargeMessageTest.java =================================================================== --- branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/client/LargeMessageTest.java 2011-11-07 04:32:51 UTC (rev 11665) +++ branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/client/LargeMessageTest.java 2011-11-08 00:45:05 UTC (rev 11666) @@ -74,6 +74,22 @@ return false; } + /** + * + */ + public LargeMessageTest() + { + super(); + } + + /** + * @param test + */ + public LargeMessageTest(String test) + { + super(test); + } + public void testRollbackPartiallyConsumedBuffer() throws Exception { for (int i = 0 ; i < 1; i++) @@ -1713,6 +1729,7 @@ 100); } + public void testPageOnLargeMessage() throws Exception { testPageOnLargeMessage(true, false); Modified: branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/client/PagingOrderTest.java =================================================================== --- branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/client/PagingOrderTest.java 2011-11-07 04:32:51 UTC (rev 11665) +++ branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/client/PagingOrderTest.java 2011-11-08 00:45:05 UTC (rev 11666) @@ -917,6 +917,8 @@ conn.close(); server.stop(); + + jmsServer.stop(); } @@ -996,6 +998,8 @@ server.stop(); + + jmsServer.stop(); } Modified: branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/client/PagingTest.java =================================================================== --- branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/client/PagingTest.java 2011-11-07 04:32:51 UTC (rev 11665) +++ branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/client/PagingTest.java 2011-11-08 00:45:05 UTC (rev 11666) @@ -121,6 +121,8 @@ protected void tearDown() throws Exception { locator.close(); + + locator = null; super.tearDown(); } @@ -263,7 +265,7 @@ PagingTest.PAGE_MAX, new HashMap<String, AddressSettings>()); server.start(); - + waitForServer(server); queue = server.locateQueue(ADDRESS); @@ -647,8 +649,7 @@ PagingTest.PAGE_MAX, new HashMap<String, AddressSettings>()); server.start(); - - + locator = createInVMNonHALocator(); locator.setBlockOnNonDurableSend(true); locator.setBlockOnDurableSend(true); @@ -662,7 +663,7 @@ session = sf.createSession(false, false, false); producer = session.createProducer(PagingTest.ADDRESS); - + for (int i = 0; i < numberOfMessages; i++) { message = session.createMessage(true); @@ -679,9 +680,9 @@ session.commit(); } } - + session.commit(); - + server.stop(); server = createServer(true, @@ -696,7 +697,7 @@ queue = server.locateQueue(ADDRESS); - // assertEquals(numberOfMessages, queue.getMessageCount()); + // assertEquals(numberOfMessages, queue.getMessageCount()); xids = new LinkedList<Xid>(); @@ -728,7 +729,6 @@ sessionConsumer.close(); - } finally { @@ -1041,50 +1041,56 @@ server.start(); - ServerLocator locator = createInVMNonHALocator(); + try + { - locator.setBlockOnNonDurableSend(true); - locator.setBlockOnDurableSend(true); - locator.setBlockOnAcknowledge(true); + ServerLocator locator = createInVMNonHALocator(); - ClientSessionFactory csf = locator.createSessionFactory(); + locator.setBlockOnNonDurableSend(true); + locator.setBlockOnDurableSend(true); + locator.setBlockOnAcknowledge(true); - ClientSession session = csf.createSession(); + ClientSessionFactory csf = locator.createSessionFactory(); - session.start(); + ClientSession session = csf.createSession(); - for (int i = 1; i <= 2; i++) - { - ClientConsumer cons = session.createConsumer("q" + i); + session.start(); - for (int j = 3; j < 6; j++) + for (int i = 1; i <= 2; i++) { - ClientMessage msg = cons.receive(5000); + ClientConsumer cons = session.createConsumer("q" + i); - assertNotNull(msg); + for (int j = 3; j < 6; j++) + { + ClientMessage msg = cons.receive(5000); - assertEquals("str-" + j, msg.getStringProperty("id")); + assertNotNull(msg); - msg.acknowledge(); + assertEquals("str-" + j, msg.getStringProperty("id")); + + msg.acknowledge(); + } + + session.commit(); + assertNull(cons.receive(500)); + } - session.commit(); - assertNull(cons.receive(500)); + session.close(); - } + long timeout = System.currentTimeMillis() + 5000; - session.close(); + while (System.currentTimeMillis() < timeout && server.getPagingManager().getPageStore(ADDRESS).isPaging()) + { + Thread.sleep(100); + } - long timeout = System.currentTimeMillis() + 5000; - - while (System.currentTimeMillis() < timeout && server.getPagingManager().getPageStore(ADDRESS).isPaging()) + locator.close(); + } + finally { - Thread.sleep(100); + server.stop(); } - - locator.close(); - - server.stop(); } public void testTwoQueuesOneNoRouting() throws Exception @@ -1277,27 +1283,28 @@ { bb.put(getSamplebyte(j)); } - + final AtomicBoolean running = new AtomicBoolean(true); - + class TCount extends Thread { Queue queue; - + TCount(Queue queue) { this.queue = queue; } + public void run() { try { while (running.get()) { - // log.info("Message count = " + queue.getMessageCount() + " on queue " + queue.getName()); + // log.info("Message count = " + queue.getMessageCount() + " on queue " + queue.getName()); queue.getMessagesAdded(); queue.getMessageCount(); - //log.info("Message added = " + queue.getMessagesAdded() + " on queue " + queue.getName()); + // log.info("Message added = " + queue.getMessagesAdded() + " on queue " + queue.getName()); Thread.sleep(10); } } @@ -1307,10 +1314,9 @@ } } }; - + TCount tcount1 = null; TCount tcount2 = null; - try { @@ -1337,8 +1343,7 @@ session.createQueue(PagingTest.ADDRESS.toString(), PagingTest.ADDRESS + "-2", null, true); } - - + ClientProducer producer = session.createProducer(PagingTest.ADDRESS); ClientMessage message = null; @@ -1377,21 +1382,21 @@ PagingTest.PAGE_MAX, new HashMap<String, AddressSettings>()); server.start(); - + Queue queue1 = server.locateQueue(PagingTest.ADDRESS.concat("-1")); - + Queue queue2 = server.locateQueue(PagingTest.ADDRESS.concat("-2")); - + assertNotNull(queue1); - + assertNotNull(queue2); - + assertNotSame(queue1, queue2); tcount1 = new TCount(queue1); - + tcount2 = new TCount(queue2); - + tcount1.start(); tcount2.start(); @@ -1500,19 +1505,19 @@ finally { running.set(false); - + if (tcount1 != null) { tcount1.interrupt(); tcount1.join(); } - + if (tcount2 != null) { tcount2.interrupt(); tcount2.join(); } - + try { server.stop(); @@ -2510,7 +2515,7 @@ producerThread.start(); - assertTrue(ready.await(10, TimeUnit.SECONDS)); + assertTrue(ready.await(100, TimeUnit.SECONDS)); ClientConsumer consumer = session.createConsumer(PagingTest.ADDRESS); Modified: branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/cluster/bridge/BridgeReconnectTest.java =================================================================== --- branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/cluster/bridge/BridgeReconnectTest.java 2011-11-07 04:32:51 UTC (rev 11665) +++ branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/cluster/bridge/BridgeReconnectTest.java 2011-11-08 00:45:05 UTC (rev 11666) @@ -124,6 +124,7 @@ forwardAddress, null, null, + HornetQClient.DEFAULT_MIN_LARGE_MESSAGE_SIZE, HornetQClient.DEFAULT_CLIENT_FAILURE_CHECK_PERIOD, HornetQClient.DEFAULT_CONNECTION_TTL, retryInterval, @@ -268,6 +269,7 @@ forwardAddress, null, null, + HornetQClient.DEFAULT_MIN_LARGE_MESSAGE_SIZE, HornetQClient.DEFAULT_CLIENT_FAILURE_CHECK_PERIOD, HornetQClient.DEFAULT_CONNECTION_TTL, retryInterval, @@ -396,6 +398,7 @@ forwardAddress, null, null, + HornetQClient.DEFAULT_MIN_LARGE_MESSAGE_SIZE, HornetQClient.DEFAULT_CLIENT_FAILURE_CHECK_PERIOD, HornetQClient.DEFAULT_CONNECTION_TTL, retryInterval, @@ -540,6 +543,7 @@ forwardAddress, null, null, + HornetQClient.DEFAULT_MIN_LARGE_MESSAGE_SIZE, clientFailureCheckPeriod, HornetQClient.DEFAULT_CONNECTION_TTL, retryInterval, @@ -679,6 +683,7 @@ forwardAddress, null, null, + HornetQClient.DEFAULT_MIN_LARGE_MESSAGE_SIZE, HornetQClient.DEFAULT_CLIENT_FAILURE_CHECK_PERIOD, HornetQClient.DEFAULT_CONNECTION_TTL, retryInterval, Modified: branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/cluster/bridge/BridgeStartTest.java =================================================================== --- branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/cluster/bridge/BridgeStartTest.java 2011-11-07 04:32:51 UTC (rev 11665) +++ branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/cluster/bridge/BridgeStartTest.java 2011-11-08 00:45:05 UTC (rev 11666) @@ -109,6 +109,7 @@ forwardAddress, null, null, + HornetQClient.DEFAULT_MIN_LARGE_MESSAGE_SIZE, HornetQClient.DEFAULT_CLIENT_FAILURE_CHECK_PERIOD, HornetQClient.DEFAULT_CONNECTION_TTL, 1000, @@ -279,6 +280,7 @@ forwardAddress, null, null, + HornetQClient.DEFAULT_MIN_LARGE_MESSAGE_SIZE, HornetQClient.DEFAULT_CLIENT_FAILURE_CHECK_PERIOD, HornetQClient.DEFAULT_CONNECTION_TTL, 500, @@ -496,6 +498,7 @@ forwardAddress, null, null, + HornetQClient.DEFAULT_MIN_LARGE_MESSAGE_SIZE, HornetQClient.DEFAULT_CLIENT_FAILURE_CHECK_PERIOD, HornetQClient.DEFAULT_CONNECTION_TTL, 1000, @@ -652,6 +655,7 @@ forwardAddress, null, null, + HornetQClient.DEFAULT_MIN_LARGE_MESSAGE_SIZE, HornetQClient.DEFAULT_CLIENT_FAILURE_CHECK_PERIOD, HornetQClient.DEFAULT_CONNECTION_TTL, 1000, Modified: branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/cluster/bridge/BridgeTest.java =================================================================== --- branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/cluster/bridge/BridgeTest.java 2011-11-07 04:32:51 UTC (rev 11665) +++ branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/cluster/bridge/BridgeTest.java 2011-11-08 00:45:05 UTC (rev 11666) @@ -141,6 +141,7 @@ forwardAddress, null, null, + HornetQClient.DEFAULT_MIN_LARGE_MESSAGE_SIZE, HornetQClient.DEFAULT_CLIENT_FAILURE_CHECK_PERIOD, HornetQClient.DEFAULT_CONNECTION_TTL, 1000, @@ -349,6 +350,7 @@ forwardAddress, filterString, null, + HornetQClient.DEFAULT_MIN_LARGE_MESSAGE_SIZE, HornetQClient.DEFAULT_CLIENT_FAILURE_CHECK_PERIOD, HornetQClient.DEFAULT_CONNECTION_TTL, 1000, @@ -538,6 +540,7 @@ forwardAddress, null, null, + HornetQClient.DEFAULT_MIN_LARGE_MESSAGE_SIZE, HornetQClient.DEFAULT_CLIENT_FAILURE_CHECK_PERIOD, HornetQClient.DEFAULT_CONNECTION_TTL, 100, @@ -698,6 +701,7 @@ forwardAddress, null, null, + HornetQClient.DEFAULT_MIN_LARGE_MESSAGE_SIZE, HornetQClient.DEFAULT_CLIENT_FAILURE_CHECK_PERIOD, HornetQClient.DEFAULT_CONNECTION_TTL, 100, @@ -891,6 +895,7 @@ forwardAddress, null, SimpleTransformer.class.getName(), + HornetQClient.DEFAULT_MIN_LARGE_MESSAGE_SIZE, HornetQClient.DEFAULT_CLIENT_FAILURE_CHECK_PERIOD, HornetQClient.DEFAULT_CONNECTION_TTL, 1000, @@ -1044,6 +1049,7 @@ forwardAddress, null, null, + HornetQClient.DEFAULT_MIN_LARGE_MESSAGE_SIZE, HornetQClient.DEFAULT_CLIENT_FAILURE_CHECK_PERIOD, HornetQClient.DEFAULT_CONNECTION_TTL, 1000, @@ -1290,6 +1296,7 @@ forwardAddress, null, null, + HornetQClient.DEFAULT_MIN_LARGE_MESSAGE_SIZE, HornetQClient.DEFAULT_CLIENT_FAILURE_CHECK_PERIOD, HornetQClient.DEFAULT_CONNECTION_TTL, 1000, @@ -1441,6 +1448,7 @@ forwardAddress, null, null, + HornetQClient.DEFAULT_MIN_LARGE_MESSAGE_SIZE, HornetQClient.DEFAULT_CLIENT_FAILURE_CHECK_PERIOD, HornetQClient.DEFAULT_CONNECTION_TTL, 1000, @@ -1601,6 +1609,7 @@ // address null, null, + HornetQClient.DEFAULT_MIN_LARGE_MESSAGE_SIZE, HornetQClient.DEFAULT_CLIENT_FAILURE_CHECK_PERIOD, HornetQClient.DEFAULT_CONNECTION_TTL, 1000, Modified: branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/cluster/bridge/BridgeWithDiscoveryGroupStartTest.java =================================================================== --- branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/cluster/bridge/BridgeWithDiscoveryGroupStartTest.java 2011-11-07 04:32:51 UTC (rev 11665) +++ branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/cluster/bridge/BridgeWithDiscoveryGroupStartTest.java 2011-11-08 00:45:05 UTC (rev 11666) @@ -116,6 +116,7 @@ forwardAddress, null, null, + HornetQClient.DEFAULT_MIN_LARGE_MESSAGE_SIZE, HornetQClient.DEFAULT_CLIENT_FAILURE_CHECK_PERIOD, HornetQClient.DEFAULT_CONNECTION_TTL, 1000, Modified: branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/cluster/bridge/BridgeWithPagingTest.java =================================================================== --- branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/cluster/bridge/BridgeWithPagingTest.java 2011-11-07 04:32:51 UTC (rev 11665) +++ branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/cluster/bridge/BridgeWithPagingTest.java 2011-11-08 00:45:05 UTC (rev 11666) @@ -113,6 +113,7 @@ forwardAddress, null, null, + HornetQClient.DEFAULT_MIN_LARGE_MESSAGE_SIZE, HornetQClient.DEFAULT_CLIENT_FAILURE_CHECK_PERIOD, HornetQClient.DEFAULT_CONNECTION_TTL, retryInterval, Modified: branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/cluster/distribution/ClusterTestBase.java =================================================================== --- branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/cluster/distribution/ClusterTestBase.java 2011-11-07 04:32:51 UTC (rev 11665) +++ branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/cluster/distribution/ClusterTestBase.java 2011-11-08 00:45:05 UTC (rev 11666) @@ -1921,6 +1921,7 @@ ClusterConnectionConfiguration clusterConf = new ClusterConnectionConfiguration(name, address, connectorFrom.getName(), + HornetQClient.DEFAULT_MIN_LARGE_MESSAGE_SIZE, ConfigurationImpl.DEFAULT_CLUSTER_FAILURE_CHECK_PERIOD, ConfigurationImpl.DEFAULT_CLUSTER_CONNECTION_TTL, retryInterval, Modified: branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/cluster/distribution/ClusteredRequestResponseTest.java =================================================================== --- branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/cluster/distribution/ClusteredRequestResponseTest.java 2011-11-07 04:32:51 UTC (rev 11665) +++ branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/cluster/distribution/ClusteredRequestResponseTest.java 2011-11-08 00:45:05 UTC (rev 11666) @@ -96,8 +96,10 @@ /* * Don't wait for the response queue bindings to get to the other side + * + * TODO: I believe this test is invalid. I'm just ignoring it for now. It will probably go away */ - public void testRequestResponseNoWaitForBindings() throws Exception + public void invalidTest_testRequestResponseNoWaitForBindings() throws Exception { setupCluster(); Modified: branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/cluster/distribution/MessageRedistributionTest.java =================================================================== --- branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/cluster/distribution/MessageRedistributionTest.java 2011-11-07 04:32:51 UTC (rev 11665) +++ branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/cluster/distribution/MessageRedistributionTest.java 2011-11-08 00:45:05 UTC (rev 11666) @@ -494,10 +494,11 @@ { for (int i = 0; i < 10; i++) { + log.info("#test " + i); setupCluster(false); startServers(0, 1, 2); - + setupSessionFactory(0, isNetty()); setupSessionFactory(1, isNetty()); setupSessionFactory(2, isNetty()); Modified: branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/cluster/distribution/TwoWayTwoNodeClusterTest.java =================================================================== --- branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/cluster/distribution/TwoWayTwoNodeClusterTest.java 2011-11-07 04:32:51 UTC (rev 11665) +++ branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/cluster/distribution/TwoWayTwoNodeClusterTest.java 2011-11-08 00:45:05 UTC (rev 11666) @@ -141,7 +141,6 @@ for (int i = 0; i < 10; i++) { - Thread.sleep(10); log.info("Sleep #test " + i); log.info("#stop #test #" + i); stopServers(1); Modified: branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/cluster/topology/HAClientTopologyWithDiscoveryTest.java =================================================================== --- branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/cluster/topology/HAClientTopologyWithDiscoveryTest.java 2011-11-07 04:32:51 UTC (rev 11665) +++ branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/cluster/topology/HAClientTopologyWithDiscoveryTest.java 2011-11-08 00:45:05 UTC (rev 11666) @@ -33,7 +33,13 @@ { return false; } + + protected boolean isFileStorage() + { + return false; + } + protected void setupCluster() throws Exception { setupCluster(false); Modified: branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/cluster/topology/IsolatedTopologyTest.java =================================================================== --- branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/cluster/topology/IsolatedTopologyTest.java 2011-11-07 04:32:51 UTC (rev 11665) +++ branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/cluster/topology/IsolatedTopologyTest.java 2011-11-08 00:45:05 UTC (rev 11666) @@ -28,7 +28,6 @@ import org.hornetq.core.remoting.impl.netty.TransportConstants; import org.hornetq.core.server.HornetQServer; import org.hornetq.tests.util.ServiceTestBase; -import org.hornetq.tests.util.UnitTestCase; /** * A IsolatedTopologyTest Modified: branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/cluster/topology/TopologyClusterTestBase.java =================================================================== --- branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/cluster/topology/TopologyClusterTestBase.java 2011-11-07 04:32:51 UTC (rev 11665) +++ branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/cluster/topology/TopologyClusterTestBase.java 2011-11-08 00:45:05 UTC (rev 11666) @@ -17,6 +17,7 @@ import java.util.ArrayList; import java.util.List; +import java.util.Map; import java.util.concurrent.CountDownLatch; import org.hornetq.api.core.HornetQException; @@ -70,6 +71,11 @@ abstract protected boolean isNetty() throws Exception; + protected boolean isFileStorage() + { + return false; + } + @Override protected void setUp() throws Exception { @@ -114,11 +120,12 @@ { ok = (ok && actual.contains(nodeIDs[expected[i]])); } - if (ok) + if (ok) { return; } - } while(System.currentTimeMillis() - start < 5000); + } + while (System.currentTimeMillis() - start < 5000); fail("did not contain all expected node ID: " + actual); } @@ -145,8 +152,8 @@ { if (e.getCode() == HornetQException.OBJECT_CLOSED || e.getCode() == HornetQException.UNBLOCKED) { - ClientSessionFactory sf = locator.createSessionFactory(); - return sf.createSession(); + ClientSessionFactory sf = locator.createSessionFactory(); + return sf.createSession(); } else { @@ -174,7 +181,14 @@ for (ClusterConnection clusterConn : clusterManager.getClusterConnections()) { - nodesCount += clusterConn.getNodes().size(); + Map<String, String> nodes = clusterConn.getNodes(); + for (String id : nodes.keySet()) + { + if (clusterConn.isNodeActive(id)) + { + nodesCount++; + } + } } if (nodesCount == count) @@ -185,85 +199,92 @@ Thread.sleep(10); } while (System.currentTimeMillis() - start < WAIT_TIMEOUT); - + log.error(clusterDescription(servers[node])); throw new IllegalStateException("Timed out waiting for cluster connections "); } + public void testReceiveNotificationsWhenOtherNodesAreStartedAndStopped() throws Throwable { startServers(0); ServerLocator locator = createHAServerLocator(); - - ((ServerLocatorImpl)locator).getTopology().setOwner("testReceive"); - final List<String> nodes = new ArrayList<String>(); - final CountDownLatch upLatch = new CountDownLatch(5); - final CountDownLatch downLatch = new CountDownLatch(4); + try + { + ((ServerLocatorImpl)locator).getTopology().setOwner("testReceive"); - locator.addClusterTopologyListener(new ClusterTopologyListener() - { - public void nodeUP(final long uniqueEventID, - String nodeID, - Pair<TransportConfiguration, TransportConfiguration> connectorPair, - boolean last) + final List<String> nodes = new ArrayList<String>(); + final CountDownLatch upLatch = new CountDownLatch(5); + final CountDownLatch downLatch = new CountDownLatch(4); + + locator.addClusterTopologyListener(new ClusterTopologyListener() { - if(!nodes.contains(nodeID)) + public void nodeUP(final long uniqueEventID, + String nodeID, + Pair<TransportConfiguration, TransportConfiguration> connectorPair, + boolean last) { - System.out.println("Node UP " + nodeID + " added"); - log.info("Node UP " + nodeID + " added"); - nodes.add(nodeID); - upLatch.countDown(); + if (!nodes.contains(nodeID)) + { + System.out.println("Node UP " + nodeID + " added"); + log.info("Node UP " + nodeID + " added"); + nodes.add(nodeID); + upLatch.countDown(); + } + else + { + System.out.println("Node UP " + nodeID + " was already here"); + log.info("Node UP " + nodeID + " was already here"); + } } - else - { - System.out.println("Node UP " + nodeID + " was already here"); - log.info("Node UP " + nodeID + " was already here"); - } - } - public void nodeDown(final long uniqueEventID, String nodeID) - { - if (nodes.contains(nodeID)) + public void nodeDown(final long uniqueEventID, String nodeID) { - log.info("Node down " + nodeID + " accepted"); - System.out.println("Node down " + nodeID + " accepted"); - nodes.remove(nodeID); - downLatch.countDown(); + if (nodes.contains(nodeID)) + { + log.info("Node down " + nodeID + " accepted"); + System.out.println("Node down " + nodeID + " accepted"); + nodes.remove(nodeID); + downLatch.countDown(); + } + else + { + log.info("Node down " + nodeID + " already removed"); + System.out.println("Node down " + nodeID + " already removed"); + } } - else - { - log.info("Node down " + nodeID + " already removed"); - System.out.println("Node down " + nodeID + " already removed"); - } - } - }); + }); - ClientSessionFactory sf = locator.createSessionFactory(); + ClientSessionFactory sf = locator.createSessionFactory(); - startServers(1, 4, 3, 2); - String[] nodeIDs = getNodeIDs(0, 1, 2, 3, 4); + startServers(1, 4, 3, 2); + String[] nodeIDs = getNodeIDs(0, 1, 2, 3, 4); - assertTrue("Was not notified that all servers are UP", upLatch.await(10, SECONDS)); - checkContains(new int[] { 0, 1, 4, 3, 2 }, nodeIDs, nodes); + assertTrue("Was not notified that all servers are UP", upLatch.await(10, SECONDS)); + checkContains(new int[] { 0, 1, 4, 3, 2 }, nodeIDs, nodes); - waitForClusterConnections(0, 4); - waitForClusterConnections(1, 4); - waitForClusterConnections(2, 4); - waitForClusterConnections(3, 4); - waitForClusterConnections(4, 4); + waitForClusterConnections(0, 4); + waitForClusterConnections(1, 4); + waitForClusterConnections(2, 4); + waitForClusterConnections(3, 4); + waitForClusterConnections(4, 4); - stopServers(2, 3, 1, 4); + stopServers(2, 3, 1, 4); - assertTrue("Was not notified that all servers are DOWN", downLatch.await(10, SECONDS)); - checkContains(new int[] { 0 }, nodeIDs, nodes); + assertTrue("Was not notified that all servers are DOWN", downLatch.await(10, SECONDS)); + checkContains(new int[] { 0 }, nodeIDs, nodes); - sf.close(); - - locator.close(); - - stopServers(0); + sf.close(); + } + finally + { + locator.close(); + + stopServers(0); + } + } public void testReceiveNotifications() throws Throwable @@ -273,73 +294,81 @@ ServerLocator locator = createHAServerLocator(); - final List<String> nodes = new ArrayList<String>(); - final CountDownLatch upLatch = new CountDownLatch(5); - final CountDownLatch downLatch = new CountDownLatch(4); + try + { - locator.addClusterTopologyListener(new ClusterTopologyListener() - { - public void nodeUP(final long uniqueEventID, - String nodeID, - Pair<TransportConfiguration, TransportConfiguration> connectorPair, - boolean last) + waitForClusterConnections(0, 4); + waitForClusterConnections(1, 4); + waitForClusterConnections(2, 4); + waitForClusterConnections(3, 4); + waitForClusterConnections(4, 4); + + final List<String> nodes = new ArrayList<String>(); + final CountDownLatch upLatch = new CountDownLatch(5); + final CountDownLatch downLatch = new CountDownLatch(4); + + locator.addClusterTopologyListener(new ClusterTopologyListener() { - if (!nodes.contains(nodeID)) + public void nodeUP(final long uniqueEventID, + String nodeID, + Pair<TransportConfiguration, TransportConfiguration> connectorPair, + boolean last) { - nodes.add(nodeID); - upLatch.countDown(); + if (!nodes.contains(nodeID)) + { + nodes.add(nodeID); + upLatch.countDown(); + } } - } - public void nodeDown(final long uniqueEventID, String nodeID) - { - if (nodes.contains(nodeID)) + public void nodeDown(final long uniqueEventID, String nodeID) { - nodes.remove(nodeID); - downLatch.countDown(); + if (nodes.contains(nodeID)) + { + nodes.remove(nodeID); + downLatch.countDown(); + } } - } - }); + }); - ClientSessionFactory sf = locator.createSessionFactory(); + ClientSessionFactory sf = locator.createSessionFactory(); - assertTrue("Was not notified that all servers are UP", upLatch.await(10, SECONDS)); - checkContains(new int[] { 0, 1, 2, 3, 4 }, nodeIDs, nodes); + assertTrue("Was not notified that all servers are UP", upLatch.await(10, SECONDS)); + checkContains(new int[] { 0, 1, 2, 3, 4 }, nodeIDs, nodes); - ClientSession session = sf.createSession(); - - waitForClusterConnections(0, 4); - waitForClusterConnections(1, 4); - waitForClusterConnections(2, 4); - waitForClusterConnections(3, 4); - waitForClusterConnections(4, 4); + ClientSession session = sf.createSession(); - stopServers(0); - session = checkSessionOrReconnect(session, locator); - checkContains(new int[] { 1, 2, 3, 4 }, nodeIDs, nodes); + stopServers(0); + session = checkSessionOrReconnect(session, locator); + checkContains(new int[] { 1, 2, 3, 4 }, nodeIDs, nodes); - stopServers(2); - session = checkSessionOrReconnect(session, locator); - checkContains(new int[] { 1, 3, 4 }, nodeIDs, nodes); + stopServers(2); + session = checkSessionOrReconnect(session, locator); + checkContains(new int[] { 1, 3, 4 }, nodeIDs, nodes); - stopServers(4); - session = checkSessionOrReconnect(session, locator); - checkContains(new int[] { 1, 3 }, nodeIDs, nodes); + stopServers(4); + session = checkSessionOrReconnect(session, locator); + checkContains(new int[] { 1, 3 }, nodeIDs, nodes); - stopServers(3); - session = checkSessionOrReconnect(session, locator); - checkContains(new int[] { 1 }, nodeIDs, nodes); + stopServers(3); + session = checkSessionOrReconnect(session, locator); + checkContains(new int[] { 1 }, nodeIDs, nodes); - stopServers(1); - - assertTrue("Was not notified that all servers are DOWN", downLatch.await(10, SECONDS)); - checkContains(new int[] {}, nodeIDs, nodes); + stopServers(1); - sf.close(); - - locator.close(); + assertTrue("Was not notified that all servers are DOWN", downLatch.await(10, SECONDS)); + checkContains(new int[] {}, nodeIDs, nodes); + + sf.close(); + } + finally + { + locator.close(); + } + } + public void testStopNodes() throws Throwable { startServers(0, 1, 2, 3, 4); @@ -347,80 +376,87 @@ ServerLocator locator = createHAServerLocator(); - final List<String> nodes = new ArrayList<String>(); - final CountDownLatch upLatch = new CountDownLatch(5); + try + { + waitForClusterConnections(0, 4); + waitForClusterConnections(1, 4); + waitForClusterConnections(2, 4); + waitForClusterConnections(3, 4); + waitForClusterConnections(4, 4); - locator.addClusterTopologyListener(new ClusterTopologyListener() - { - public void nodeUP(final long uniqueEventID, String nodeID, - Pair<TransportConfiguration, TransportConfiguration> connectorPair, - boolean last) + final List<String> nodes = new ArrayList<String>(); + final CountDownLatch upLatch = new CountDownLatch(5); + + locator.addClusterTopologyListener(new ClusterTopologyListener() { - if (!nodes.contains(nodeID)) + public void nodeUP(final long uniqueEventID, + String nodeID, + Pair<TransportConfiguration, TransportConfiguration> connectorPair, + boolean last) { - nodes.add(nodeID); - upLatch.countDown(); + if (!nodes.contains(nodeID)) + { + nodes.add(nodeID); + upLatch.countDown(); + } } - } - public void nodeDown(final long uniqueEventID, String nodeID) - { - if (nodes.contains(nodeID)) + public void nodeDown(final long uniqueEventID, String nodeID) { - nodes.remove(nodeID); + if (nodes.contains(nodeID)) + { + nodes.remove(nodeID); + } } - } - }); + }); - ClientSessionFactory sf = locator.createSessionFactory(); + ClientSessionFactory sf = locator.createSessionFactory(); - assertTrue("Was not notified that all servers are UP", upLatch.await(10, SECONDS)); - checkContains(new int[] { 0, 1, 2, 3, 4 }, nodeIDs, nodes); + assertTrue("Was not notified that all servers are UP", upLatch.await(10, SECONDS)); + checkContains(new int[] { 0, 1, 2, 3, 4 }, nodeIDs, nodes); - waitForClusterConnections(0, 4); - waitForClusterConnections(1, 4); - waitForClusterConnections(2, 4); - waitForClusterConnections(3, 4); - waitForClusterConnections(4, 4); + ClientSession session = sf.createSession(); - ClientSession session = sf.createSession(); - - stopServers(0); - assertFalse(servers[0].isStarted()); - session = checkSessionOrReconnect(session, locator); - checkContains(new int[] { 1, 2, 3, 4 }, nodeIDs, nodes); + stopServers(0); + assertFalse(servers[0].isStarted()); + session = checkSessionOrReconnect(session, locator); + checkContains(new int[] { 1, 2, 3, 4 }, nodeIDs, nodes); - stopServers(2); - assertFalse(servers[2].isStarted()); - session = checkSessionOrReconnect(session, locator); - checkContains(new int[] { 1, 3, 4 }, nodeIDs, nodes); + stopServers(2); + assertFalse(servers[2].isStarted()); + session = checkSessionOrReconnect(session, locator); + checkContains(new int[] { 1, 3, 4 }, nodeIDs, nodes); - stopServers(4); - assertFalse(servers[4].isStarted()); - session = checkSessionOrReconnect(session, locator); - checkContains(new int[] { 1, 3 }, nodeIDs, nodes); + stopServers(4); + assertFalse(servers[4].isStarted()); + session = checkSessionOrReconnect(session, locator); + checkContains(new int[] { 1, 3 }, nodeIDs, nodes); - stopServers(3); - assertFalse(servers[3].isStarted()); + stopServers(3); + assertFalse(servers[3].isStarted()); - session = checkSessionOrReconnect(session, locator); - checkContains(new int[] { 1 }, nodeIDs, nodes); + session = checkSessionOrReconnect(session, locator); + checkContains(new int[] { 1 }, nodeIDs, nodes); - stopServers(1); - assertFalse(servers[1].isStarted()); - try + stopServers(1); + assertFalse(servers[1].isStarted()); + try + { + session = checkSessionOrReconnect(session, locator); + fail(); + } + catch (Exception e) + { + + } + } + finally { - session = checkSessionOrReconnect(session, locator); - fail(); + locator.close(); } - catch (Exception e) - { - } - - locator.close(); } - + public void testMultipleClientSessionFactories() throws Throwable { startServers(0, 1, 2, 3, 4); @@ -428,67 +464,75 @@ ServerLocator locator = createHAServerLocator(); - final List<String> nodes = new ArrayList<String>(); - final CountDownLatch upLatch = new CountDownLatch(5); - final CountDownLatch downLatch = new CountDownLatch(4); + try + { - locator.addClusterTopologyListener(new ClusterTopologyListener() - { - public void nodeUP(final long uniqueEventID, String nodeID, - Pair<TransportConfiguration, TransportConfiguration> connectorPair, - boolean last) + waitForClusterConnections(0, 4); + waitForClusterConnections(1, 4); + waitForClusterConnections(2, 4); + waitForClusterConnections(3, 4); + waitForClusterConnections(4, 4); + + final List<String> nodes = new ArrayList<String>(); + final CountDownLatch upLatch = new CountDownLatch(5); + final CountDownLatch downLatch = new CountDownLatch(4); + + locator.addClusterTopologyListener(new ClusterTopologyListener() { - if (!nodes.contains(nodeID)) + public void nodeUP(final long uniqueEventID, + String nodeID, + Pair<TransportConfiguration, TransportConfiguration> connectorPair, + boolean last) { - nodes.add(nodeID); - upLatch.countDown(); + if (!nodes.contains(nodeID)) + { + nodes.add(nodeID); + upLatch.countDown(); + } } - } - public void nodeDown(final long uniqueEventID, String nodeID) - { - if (nodes.contains(nodeID)) + public void nodeDown(final long uniqueEventID, String nodeID) { - nodes.remove(nodeID); - downLatch.countDown(); + if (nodes.contains(nodeID)) + { + nodes.remove(nodeID); + downLatch.countDown(); + } } - } - }); + }); - ClientSessionFactory[] sfs = new ClientSessionFactory[] { - locator.createSessionFactory(), - locator.createSessionFactory(), - locator.createSessionFactory(), - locator.createSessionFactory(), - locator.createSessionFactory() }; - assertTrue("Was not notified that all servers are UP", upLatch.await(10, SECONDS)); - checkContains(new int[] { 0, 1, 2, 3, 4 }, nodeIDs, nodes); + ClientSessionFactory[] sfs = new ClientSessionFactory[] { locator.createSessionFactory(), + locator.createSessionFactory(), + locator.createSessionFactory(), + locator.createSessionFactory(), + locator.createSessionFactory() }; + assertTrue("Was not notified that all servers are UP", upLatch.await(10, SECONDS)); + checkContains(new int[] { 0, 1, 2, 3, 4 }, nodeIDs, nodes); - waitForClusterConnections(0, 4); - waitForClusterConnections(1, 4); - waitForClusterConnections(2, 4); - waitForClusterConnections(3, 4); - waitForClusterConnections(4, 4); - //we cant close all of the servers, we need to leave one up to notify us - stopServers(4, 2, 3, 1); + // we cant close all of the servers, we need to leave one up to notify us + stopServers(4, 2, 3, 1); - boolean ok = downLatch.await(10, SECONDS); - if(!ok) - { - System.out.println("TopologyClusterTestBase.testMultipleClientSessionFactories"); + boolean ok = downLatch.await(10, SECONDS); + if (!ok) + { + log.warn("TopologyClusterTestBase.testMultipleClientSessionFactories will fail"); + } + assertTrue("Was not notified that all servers are Down", ok); + checkContains(new int[] { 0 }, nodeIDs, nodes); + + for (int i = 0; i < sfs.length; i++) + { + ClientSessionFactory sf = sfs[i]; + sf.close(); + } } - assertTrue("Was not notified that all servers are Down", ok); - checkContains(new int[] { 0 }, nodeIDs, nodes); - - for (int i = 0; i < sfs.length; i++) + finally { - ClientSessionFactory sf = sfs[i]; - sf.close(); + locator.close(); + + stopServers(0); } - - locator.close(); - stopServers(0); } // Private ------------------------------------------------------- Modified: branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/largemessage/LargeMessageTestBase.java =================================================================== --- branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/largemessage/LargeMessageTestBase.java 2011-11-07 04:32:51 UTC (rev 11665) +++ branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/largemessage/LargeMessageTestBase.java 2011-11-08 00:45:05 UTC (rev 11666) @@ -69,7 +69,18 @@ // Package protected --------------------------------------------- // Protected ----------------------------------------------------- + + public LargeMessageTestBase(String test) + { + super(test); + } + + public LargeMessageTestBase() + { + super(); + } + @Override protected void tearDown() throws Exception { Modified: branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/management/BridgeControlTest.java =================================================================== --- branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/management/BridgeControlTest.java 2011-11-07 04:32:51 UTC (rev 11665) +++ branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/management/BridgeControlTest.java 2011-11-08 00:45:05 UTC (rev 11666) @@ -166,6 +166,7 @@ targetQueueConfig.getAddress(), null, null, + HornetQClient.DEFAULT_MIN_LARGE_MESSAGE_SIZE, HornetQClient.DEFAULT_CLIENT_FAILURE_CHECK_PERIOD, HornetQClient.DEFAULT_CONNECTION_TTL, RandomUtil.randomPositiveLong(), Modified: branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/management/BridgeControlUsingCoreTest.java =================================================================== --- branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/management/BridgeControlUsingCoreTest.java 2011-11-07 04:32:51 UTC (rev 11665) +++ branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/management/BridgeControlUsingCoreTest.java 2011-11-08 00:45:05 UTC (rev 11666) @@ -144,6 +144,7 @@ targetQueueConfig.getAddress(), null, null, + HornetQClient.DEFAULT_MIN_LARGE_MESSAGE_SIZE, HornetQClient.DEFAULT_CLIENT_FAILURE_CHECK_PERIOD, HornetQClient.DEFAULT_CONNECTION_TTL, RandomUtil.randomPositiveLong(), Modified: branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/ra/HornetQMessageHandlerSecurityTest.java =================================================================== --- branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/ra/HornetQMessageHandlerSecurityTest.java 2011-11-07 04:32:51 UTC (rev 11665) +++ branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/ra/HornetQMessageHandlerSecurityTest.java 2011-11-08 00:45:05 UTC (rev 11666) @@ -22,12 +22,11 @@ import org.hornetq.core.server.impl.QueueImpl; import org.hornetq.ra.HornetQResourceAdapter; import org.hornetq.ra.inflow.HornetQActivationSpec; +import org.hornetq.tests.util.UnitTestCase; -import javax.resource.ResourceException; import java.util.HashSet; import java.util.Set; import java.util.concurrent.CountDownLatch; -import java.util.concurrent.TimeUnit; /** * @author <a href="mailto:andy.taylor@jboss.org">Andy Taylor</a> @@ -44,6 +43,7 @@ public void testSimpleMessageReceivedOnQueueWithSecurityFails() throws Exception { HornetQResourceAdapter qResourceAdapter = new HornetQResourceAdapter(); + qResourceAdapter.setConnectorClassName(UnitTestCase.INVM_CONNECTOR_FACTORY); MyBootstrapContext ctx = new MyBootstrapContext(); qResourceAdapter.start(ctx); HornetQActivationSpec spec = new HornetQActivationSpec(); @@ -74,6 +74,7 @@ roles.add(role); server.getSecurityRepository().addMatch(MDBQUEUEPREFIXED, roles); HornetQResourceAdapter qResourceAdapter = new HornetQResourceAdapter(); + qResourceAdapter.setConnectorClassName(UnitTestCase.INVM_CONNECTOR_FACTORY); MyBootstrapContext ctx = new MyBootstrapContext(); qResourceAdapter.start(ctx); HornetQActivationSpec spec = new HornetQActivationSpec(); Modified: branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/ra/HornetQMessageHandlerTest.java =================================================================== --- branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/ra/HornetQMessageHandlerTest.java 2011-11-07 04:32:51 UTC (rev 11665) +++ branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/ra/HornetQMessageHandlerTest.java 2011-11-08 00:45:05 UTC (rev 11666) @@ -20,6 +20,7 @@ import org.hornetq.core.postoffice.impl.LocalQueueBinding; import org.hornetq.ra.HornetQResourceAdapter; import org.hornetq.ra.inflow.HornetQActivationSpec; +import org.hornetq.tests.util.UnitTestCase; import javax.jms.Message; import java.util.concurrent.CountDownLatch; @@ -41,6 +42,7 @@ public void testSimpleMessageReceivedOnQueue() throws Exception { HornetQResourceAdapter qResourceAdapter = new HornetQResourceAdapter(); + qResourceAdapter.setConnectorClassName(UnitTestCase.INVM_CONNECTOR_FACTORY); MyBootstrapContext ctx = new MyBootstrapContext(); qResourceAdapter.start(ctx); HornetQActivationSpec spec = new HornetQActivationSpec(); @@ -72,6 +74,7 @@ public void testInvalidAckMode() throws Exception { HornetQResourceAdapter qResourceAdapter = new HornetQResourceAdapter(); + qResourceAdapter.setConnectorClassName(UnitTestCase.INVM_CONNECTOR_FACTORY); MyBootstrapContext ctx = new MyBootstrapContext(); qResourceAdapter.start(ctx); HornetQActivationSpec spec = new HornetQActivationSpec(); @@ -90,6 +93,7 @@ public void testSimpleMessageReceivedOnQueueInLocalTX() throws Exception { HornetQResourceAdapter qResourceAdapter = new HornetQResourceAdapter(); + qResourceAdapter.setConnectorClassName(UnitTestCase.INVM_CONNECTOR_FACTORY); qResourceAdapter.setUseLocalTx(true); MyBootstrapContext ctx = new MyBootstrapContext(); qResourceAdapter.start(ctx); @@ -126,6 +130,7 @@ public void testSimpleMessageReceivedOnQueueWithSelector() throws Exception { HornetQResourceAdapter qResourceAdapter = new HornetQResourceAdapter(); + qResourceAdapter.setConnectorClassName(UnitTestCase.INVM_CONNECTOR_FACTORY); MyBootstrapContext ctx = new MyBootstrapContext(); qResourceAdapter.start(ctx); HornetQActivationSpec spec = new HornetQActivationSpec(); @@ -162,6 +167,7 @@ public void testEndpointDeactivated() throws Exception { HornetQResourceAdapter qResourceAdapter = new HornetQResourceAdapter(); + qResourceAdapter.setConnectorClassName(UnitTestCase.INVM_CONNECTOR_FACTORY); MyBootstrapContext ctx = new MyBootstrapContext(); qResourceAdapter.start(ctx); HornetQActivationSpec spec = new HornetQActivationSpec(); @@ -185,6 +191,7 @@ public void testMaxSessions() throws Exception { HornetQResourceAdapter qResourceAdapter = new HornetQResourceAdapter(); + qResourceAdapter.setConnectorClassName(UnitTestCase.INVM_CONNECTOR_FACTORY); MyBootstrapContext ctx = new MyBootstrapContext(); qResourceAdapter.start(ctx); HornetQActivationSpec spec = new HornetQActivationSpec(); @@ -207,6 +214,7 @@ public void testSimpleTopic() throws Exception { HornetQResourceAdapter qResourceAdapter = new HornetQResourceAdapter(); + qResourceAdapter.setConnectorClassName(UnitTestCase.INVM_CONNECTOR_FACTORY); MyBootstrapContext ctx = new MyBootstrapContext(); qResourceAdapter.start(ctx); HornetQActivationSpec spec = new HornetQActivationSpec(); @@ -237,6 +245,7 @@ public void testDurableSubscription() throws Exception { HornetQResourceAdapter qResourceAdapter = new HornetQResourceAdapter(); + qResourceAdapter.setConnectorClassName(UnitTestCase.INVM_CONNECTOR_FACTORY); MyBootstrapContext ctx = new MyBootstrapContext(); qResourceAdapter.start(ctx); HornetQActivationSpec spec = new HornetQActivationSpec(); @@ -293,6 +302,7 @@ public void testNonDurableSubscription() throws Exception { HornetQResourceAdapter qResourceAdapter = new HornetQResourceAdapter(); + qResourceAdapter.setConnectorClassName(UnitTestCase.INVM_CONNECTOR_FACTORY); MyBootstrapContext ctx = new MyBootstrapContext(); qResourceAdapter.start(ctx); HornetQActivationSpec spec = new HornetQActivationSpec(); @@ -340,6 +350,7 @@ public void testSelectorChangedWithTopic() throws Exception { HornetQResourceAdapter qResourceAdapter = new HornetQResourceAdapter(); + qResourceAdapter.setConnectorClassName(UnitTestCase.INVM_CONNECTOR_FACTORY); MyBootstrapContext ctx = new MyBootstrapContext(); qResourceAdapter.start(ctx); HornetQActivationSpec spec = new HornetQActivationSpec(); @@ -396,6 +407,7 @@ public void testSelectorNotChangedWithTopic() throws Exception { HornetQResourceAdapter qResourceAdapter = new HornetQResourceAdapter(); + qResourceAdapter.setConnectorClassName(UnitTestCase.INVM_CONNECTOR_FACTORY); MyBootstrapContext ctx = new MyBootstrapContext(); qResourceAdapter.start(ctx); HornetQActivationSpec spec = new HornetQActivationSpec(); Modified: branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/ra/HornetQMessageHandlerXATest.java =================================================================== --- branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/ra/HornetQMessageHandlerXATest.java 2011-11-07 04:32:51 UTC (rev 11665) +++ branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/ra/HornetQMessageHandlerXATest.java 2011-11-08 00:45:05 UTC (rev 11666) @@ -18,6 +18,7 @@ import org.hornetq.core.transaction.impl.XidImpl; import org.hornetq.ra.HornetQResourceAdapter; import org.hornetq.ra.inflow.HornetQActivationSpec; +import org.hornetq.tests.util.UnitTestCase; import org.hornetq.utils.UUIDGenerator; import javax.resource.ResourceException; @@ -43,6 +44,7 @@ public void testXACommit() throws Exception { HornetQResourceAdapter qResourceAdapter = new HornetQResourceAdapter(); + qResourceAdapter.setConnectorClassName(UnitTestCase.INVM_CONNECTOR_FACTORY); MyBootstrapContext ctx = new MyBootstrapContext(); qResourceAdapter.start(ctx); HornetQActivationSpec spec = new HornetQActivationSpec(); @@ -74,6 +76,7 @@ public void testXARollback() throws Exception { HornetQResourceAdapter qResourceAdapter = new HornetQResourceAdapter(); + qResourceAdapter.setConnectorClassName(UnitTestCase.INVM_CONNECTOR_FACTORY); MyBootstrapContext ctx = new MyBootstrapContext(); qResourceAdapter.start(ctx); HornetQActivationSpec spec = new HornetQActivationSpec(); Modified: branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/ra/ResourceAdapterTest.java =================================================================== --- branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/ra/ResourceAdapterTest.java 2011-11-07 04:32:51 UTC (rev 11665) +++ branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/ra/ResourceAdapterTest.java 2011-11-08 00:45:05 UTC (rev 11666) @@ -16,6 +16,7 @@ import org.hornetq.ra.HornetQResourceAdapter; import org.hornetq.ra.inflow.HornetQActivation; import org.hornetq.ra.inflow.HornetQActivationSpec; +import org.hornetq.tests.util.UnitTestCase; import javax.resource.ResourceException; import javax.resource.spi.endpoint.MessageEndpoint; @@ -31,6 +32,7 @@ public void testStartStop() throws Exception { HornetQResourceAdapter qResourceAdapter = new HornetQResourceAdapter(); + qResourceAdapter.setConnectorClassName(UnitTestCase.INVM_CONNECTOR_FACTORY); HornetQRATestBase.MyBootstrapContext ctx = new HornetQRATestBase.MyBootstrapContext(); qResourceAdapter.setTransactionManagerLocatorClass(""); Modified: branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/unit/core/config/impl/FileConfigurationTest.java =================================================================== --- branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/unit/core/config/impl/FileConfigurationTest.java 2011-11-07 04:32:51 UTC (rev 11665) +++ branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/unit/core/config/impl/FileConfigurationTest.java 2011-11-08 00:45:05 UTC (rev 11666) @@ -199,6 +199,7 @@ { Assert.assertEquals("bridge1", bc.getName()); Assert.assertEquals("queue1", bc.getQueueName()); + Assert.assertEquals(4, bc.getMinLargeMessageSize()); Assert.assertEquals("bridge-forwarding-address1", bc.getForwardingAddress()); Assert.assertEquals("sku > 1", bc.getFilterString()); Assert.assertEquals("org.foo.BridgeTransformer", bc.getTransformerClassName()); @@ -227,6 +228,7 @@ if (ccc.getName().equals("cluster-connection1")) { Assert.assertEquals("cluster-connection1", ccc.getName()); + Assert.assertEquals(321, ccc.getMinLargeMessageSize()); Assert.assertEquals("queues1", ccc.getAddress()); Assert.assertEquals(3, ccc.getRetryInterval()); Assert.assertEquals(true, ccc.isDuplicateDetection()); Modified: branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/unit/core/journal/impl/fakes/FakeSequentialFileFactory.java =================================================================== --- branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/unit/core/journal/impl/fakes/FakeSequentialFileFactory.java 2011-11-07 04:32:51 UTC (rev 11665) +++ branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/unit/core/journal/impl/fakes/FakeSequentialFileFactory.java 2011-11-08 00:45:05 UTC (rev 11666) @@ -609,7 +609,7 @@ { } - public SequentialFile copy() + public SequentialFile cloneFile() { return null; // To change body of implemented methods use File | Settings | File Templates. } Modified: branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/unit/core/persistence/impl/BatchIDGeneratorUnitTest.java =================================================================== --- branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/unit/core/persistence/impl/BatchIDGeneratorUnitTest.java 2011-11-07 04:32:51 UTC (rev 11665) +++ branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/unit/core/persistence/impl/BatchIDGeneratorUnitTest.java 2011-11-08 00:45:05 UTC (rev 11666) @@ -127,6 +127,8 @@ Assert.assertEquals("No Ids were generated, so the currentID was supposed to stay the same", lastId, batch.getCurrentID()); + + journal.stop(); } Modified: branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/util/UnitTestCase.java =================================================================== --- branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/util/UnitTestCase.java 2011-11-07 04:32:51 UTC (rev 11665) +++ branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/util/UnitTestCase.java 2011-11-08 00:45:05 UTC (rev 11666) @@ -111,8 +111,23 @@ // Attributes ---------------------------------------------------- - private static final String testDir = System.getProperty("java.io.tmpdir", "/tmp") + "/hornetq-unit-test"; + private final String baseDir = System.getProperty("java.io.tmpdir", "/tmp") + File.separator + "hornetq-unit-test"; + private long timeStart = System.currentTimeMillis(); + + public long getTimeStart() + { + return timeStart; + } + + private String testDir = baseDir + File.separator + timeStart; + + public void setTimeStart(long time) + { + timeStart = time; + testDir = baseDir + File.separator + timeStart; + } + // There is a verification about thread leakages. We only fail a single thread when this happens private static Set<Thread> alreadyFailedThread = new HashSet<Thread>(); @@ -617,7 +632,7 @@ { // Need to delete the root - File file = new File(testDir); + File file = new File(baseDir); deleteDirectory(file); file.mkdirs(); @@ -627,8 +642,6 @@ recreateDirectory(getLargeMessagesDir(testDir)); recreateDirectory(getClientLargeMessagesDir(testDir)); recreateDirectory(getTemporaryDir(testDir)); - - System.out.println("deleted " + testDir); } /** @@ -899,7 +912,7 @@ OperationContextImpl.clearContext(); - deleteDirectory(new File(getTestDir())); + clearData(getTestDir()); InVMRegistry.instance.clear(); @@ -914,7 +927,10 @@ protected void tearDown() throws Exception { cleanupPools(); + + deleteDirectory(new File(baseDir)); + Map<Thread, StackTraceElement[]> threadMap = Thread.getAllStackTraces(); for (Thread thread : threadMap.keySet()) { @@ -1162,16 +1178,43 @@ protected boolean deleteDirectory(final File directory) { + if (!directory.exists()) + { + return true; + } + else if (directory.isDirectory()) { String[] files = directory.list(); for (int j = 0; j < files.length; j++) { - if (!deleteDirectory(new File(directory, files[j]))) + try { - return false; + + File fileTmp = new File(directory, files[j]); + if (!deleteDirectory(fileTmp)) + { + // This is because of Windows is dumb on releasing files + log.warn("could not delete " + fileTmp); + forceGC(); + if (!deleteDirectory(fileTmp)) + { + log.warn("**************************************************************"); + log.warn("could not delete " + fileTmp + " even afer a retry on GC"); + log.warn("**************************************************************"); + } + else + { + log.info(fileTmp + " was deleted without a problem after a retry on GC"); + } + return false; + } } + catch (Throwable e) + { + e.printStackTrace(); + } } }