[hornetq-commits] JBoss hornetq SVN: r7979 - in branches/Branch_Replication_Changes: src/main/org/hornetq/core/client/impl and 12 other directories.

Author: timfox Date: 2009-09-22 11:08:41 -0400 (Tue, 22 Sep 2009) New Revision: 7979 Added: branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/integration/cluster/failover/AsynchronousFailoverTest.java branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/integration/cluster/failover/DelayInterceptor2.java branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/integration/cluster/failover/DelayInterceptor3.java branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/integration/cluster/failover/FailoverTestBase.java branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/integration/cluster/failover/NettyAsynchronousFailoverTest.java branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/integration/cluster/failover/NettyFailoverTest.java branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/integration/cluster/reattach/ branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/integration/cluster/reattach/MultiThreadRandomReattachTest.java branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/integration/cluster/reattach/MultiThreadRandomReattachTestBase.java branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/integration/cluster/reattach/MultiThreadReattachSupport.java branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/integration/cluster/reattach/NettyMultiThreadRandomReattachTest.java branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/integration/cluster/reattach/RandomReattachTest.java branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/integration/cluster/reattach/ReattachTest.java Removed: branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/integration/cluster/failover/FailoverNoSessionsFailoverTest.java branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/integration/cluster/failover/FailureListenerOnFailoverTest.java branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/integration/cluster/failover/MultiThreadFailoverSupport.java branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/integration/cluster/failover/MultiThreadRandomReattachTest.java branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/integration/cluster/failover/MultiThreadRandomReattachTestBase.java branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/integration/cluster/failover/NettyMultiThreadRandomReattachTest.java branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/integration/cluster/failover/RandomReattachTest.java branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/integration/cluster/failover/ReattachTest.java Modified: branches/Branch_Replication_Changes/src/main/org/hornetq/core/client/ClientSessionFactory.java branches/Branch_Replication_Changes/src/main/org/hornetq/core/client/impl/ClientProducerImpl.java branches/Branch_Replication_Changes/src/main/org/hornetq/core/client/impl/ClientSessionFactoryImpl.java branches/Branch_Replication_Changes/src/main/org/hornetq/core/client/impl/ClientSessionImpl.java branches/Branch_Replication_Changes/src/main/org/hornetq/core/client/impl/ClientSessionInternal.java branches/Branch_Replication_Changes/src/main/org/hornetq/core/client/impl/ConnectionManagerImpl.java branches/Branch_Replication_Changes/src/main/org/hornetq/core/client/impl/DelegatingSession.java branches/Branch_Replication_Changes/src/main/org/hornetq/core/remoting/impl/RemotingConnectionImpl.java branches/Branch_Replication_Changes/src/main/org/hornetq/core/remoting/server/impl/RemotingServiceImpl.java branches/Branch_Replication_Changes/src/main/org/hornetq/core/server/cluster/impl/BridgeImpl.java branches/Branch_Replication_Changes/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java branches/Branch_Replication_Changes/src/main/org/hornetq/integration/transports/netty/VirtualExecutorService.java branches/Branch_Replication_Changes/tests/jms-tests/src/org/hornetq/jms/tests/message/MessageHeaderTest.java branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/integration/clientcrash/DummyInterceptor.java branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/integration/clientcrash/DummyInterceptorB.java branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/integration/cluster/failover/FailoverTest.java branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/soak/failover/RandomFailoverSoakTest.java branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/stress/failover/MultiThreadRandomFailoverStressTest.java branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/stress/failover/RandomFailoverStressTest.java Log: replication Modified: branches/Branch_Replication_Changes/src/main/org/hornetq/core/client/ClientSessionFactory.java =================================================================== --- branches/Branch_Replication_Changes/src/main/org/hornetq/core/client/ClientSessionFactory.java 2009-09-22 13:57:55 UTC (rev 7978) +++ branches/Branch_Replication_Changes/src/main/org/hornetq/core/client/ClientSessionFactory.java 2009-09-22 15:08:41 UTC (rev 7979) @@ -17,6 +17,7 @@ import org.hornetq.core.config.TransportConfiguration; import org.hornetq.core.exception.HornetQException; +import org.hornetq.core.remoting.Interceptor; import org.hornetq.utils.Pair; /** @@ -144,6 +145,10 @@ int getReconnectAttempts(); void setReconnectAttempts(int reconnectAttempts); + + boolean isUseReattach(); + + void setUseReattach(boolean reattach); boolean isFailoverOnServerShutdown(); @@ -164,6 +169,10 @@ long getDiscoveryRefreshTimeout(); void setDiscoveryRefreshTimeout(long discoveryRefreshTimeout); + + void addInterceptor(Interceptor interceptor); + boolean removeInterceptor(Interceptor interceptor); + void close(); } Modified: branches/Branch_Replication_Changes/src/main/org/hornetq/core/client/impl/ClientProducerImpl.java =================================================================== --- branches/Branch_Replication_Changes/src/main/org/hornetq/core/client/impl/ClientProducerImpl.java 2009-09-22 13:57:55 UTC (rev 7978) +++ branches/Branch_Replication_Changes/src/main/org/hornetq/core/client/impl/ClientProducerImpl.java 2009-09-22 15:08:41 UTC (rev 7979) @@ -224,6 +224,7 @@ } else if (sendBlocking) { + //log.info("sending blocking"); channel.sendBlocking(message); } else Modified: branches/Branch_Replication_Changes/src/main/org/hornetq/core/client/impl/ClientSessionFactoryImpl.java =================================================================== --- branches/Branch_Replication_Changes/src/main/org/hornetq/core/client/impl/ClientSessionFactoryImpl.java 2009-09-22 13:57:55 UTC (rev 7978) +++ branches/Branch_Replication_Changes/src/main/org/hornetq/core/client/impl/ClientSessionFactoryImpl.java 2009-09-22 15:08:41 UTC (rev 7979) @@ -21,6 +21,7 @@ import java.util.List; import java.util.Map; import java.util.Set; +import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; @@ -36,6 +37,7 @@ import org.hornetq.core.config.TransportConfiguration; import org.hornetq.core.exception.HornetQException; import org.hornetq.core.logging.Logger; +import org.hornetq.core.remoting.Interceptor; import org.hornetq.utils.HornetQThreadFactory; import org.hornetq.utils.Pair; import org.hornetq.utils.UUIDGenerator; @@ -101,6 +103,8 @@ public static final double DEFAULT_RETRY_INTERVAL_MULTIPLIER = 1d; public static final int DEFAULT_RECONNECT_ATTEMPTS = 0; + + public static final boolean DEFAULT_USE_REATTACH = false; public static final boolean DEFAULT_FAILOVER_ON_SERVER_SHUTDOWN = false; @@ -188,10 +192,14 @@ private double retryIntervalMultiplier; private int reconnectAttempts; + + private boolean useReattach; private volatile boolean closed; private boolean failoverOnServerShutdown; + + private final List<Interceptor> interceptors = new CopyOnWriteArrayList<Interceptor>(); private static ExecutorService globalThreadPool; @@ -284,8 +292,10 @@ retryInterval, retryIntervalMultiplier, reconnectAttempts, + useReattach, threadPool, - scheduledThreadPool); + scheduledThreadPool, + interceptors); connectionManagerMap.put(pair, cm); } @@ -353,6 +363,8 @@ retryIntervalMultiplier = DEFAULT_RETRY_INTERVAL_MULTIPLIER; reconnectAttempts = DEFAULT_RECONNECT_ATTEMPTS; + + useReattach = DEFAULT_USE_REATTACH; failoverOnServerShutdown = DEFAULT_FAILOVER_ON_SERVER_SHUTDOWN; } @@ -654,7 +666,18 @@ checkWrite(); this.reconnectAttempts = reconnectAttempts; } + + public synchronized boolean isUseReattach() + { + return useReattach; + } + public synchronized void setUseReattach(boolean reattach) + { + checkWrite(); + this.useReattach = reattach; + } + public synchronized boolean isFailoverOnServerShutdown() { return failoverOnServerShutdown; @@ -703,7 +726,17 @@ { return discoveryRefreshTimeout; } + + public void addInterceptor(final Interceptor interceptor) + { + interceptors.add(interceptor); + } + public boolean removeInterceptor(final Interceptor interceptor) + { + return interceptors.remove(interceptor); + } + public synchronized void setDiscoveryRefreshTimeout(long discoveryRefreshTimeout) { checkWrite(); @@ -907,8 +940,10 @@ retryInterval, retryIntervalMultiplier, reconnectAttempts, + useReattach, threadPool, - scheduledThreadPool); + scheduledThreadPool, + interceptors); connectionManagerMap.put(connectorPair, connectionManager); } Modified: branches/Branch_Replication_Changes/src/main/org/hornetq/core/client/impl/ClientSessionImpl.java =================================================================== --- branches/Branch_Replication_Changes/src/main/org/hornetq/core/client/impl/ClientSessionImpl.java 2009-09-22 13:57:55 UTC (rev 7978) +++ branches/Branch_Replication_Changes/src/main/org/hornetq/core/client/impl/ClientSessionImpl.java 2009-09-22 15:08:41 UTC (rev 7979) @@ -42,6 +42,8 @@ import org.hornetq.core.remoting.impl.wireformat.CreateSessionMessage; import org.hornetq.core.remoting.impl.wireformat.CreateSessionResponseMessage; import org.hornetq.core.remoting.impl.wireformat.PacketImpl; +import org.hornetq.core.remoting.impl.wireformat.ReattachSessionMessage; +import org.hornetq.core.remoting.impl.wireformat.ReattachSessionResponseMessage; import org.hornetq.core.remoting.impl.wireformat.RollbackMessage; import org.hornetq.core.remoting.impl.wireformat.SessionAcknowledgeMessage; import org.hornetq.core.remoting.impl.wireformat.SessionBindingQueryMessage; @@ -167,7 +169,7 @@ private SendAcknowledgementHandler sendAckHandler; - // private volatile boolean closedSent; + private volatile boolean closedSent; private volatile boolean rollbackOnly; @@ -743,7 +745,7 @@ { closeChildren(); - // closedSent = true; + closedSent = true; channel.sendBlocking(new SessionCloseMessage()); } @@ -774,76 +776,78 @@ sendAckHandler = handler; } - // // Needs to be synchronized to prevent issues with occurring concurrently with close() - // public synchronized boolean handleOldFailover(final RemotingConnection backupConnection) - // { - // if (closed) - // { - // return true; - // } - // - // boolean ok = false; - // - // // We lock the channel to prevent any packets to be added to the resend - // // cache during the failover process - // channel.lock(); - // - // try - // { - // channel.transferConnection(backupConnection); - // - // backupConnection.syncIDGeneratorSequence(remotingConnection.getIDGeneratorSequence()); - // - // remotingConnection = backupConnection; - // - // Packet request = new ReattachSessionMessage(name, channel.getLastReceivedCommandID()); - // - // Channel channel1 = backupConnection.getChannel(1, -1, false); - // - // ReattachSessionResponseMessage response = (ReattachSessionResponseMessage)channel1.sendBlocking(request); - // - // if (response.isSessionFound()) - // { - // channel.replayCommands(response.getLastReceivedCommandID(), channel.getID()); - // - // ok = true; - // } - // else - // { - // if (closedSent) - // { - // // a session re-attach may fail, if the session close was sent before failover started, hit the server, - // // processed, then before the response was received back, failover occurred, re-attach was attempted. in - // // this case it's ok - we don't want to call any failure listeners and we don't want to halt the rest of - // // the failover process. - // // - // // however if session re-attach fails and the session was not in a call to close, then we DO want to call - // // the session listeners so we return false - // // - // // Also session reattach will fail if the server is restarted - so the session is lost - // ok = true; - // } - // else - // { - // log.warn(System.identityHashCode(this) + " Session not found on server when attempting to re-attach"); - // } - // - // channel.returnBlocking(); - // } - // - // } - // catch (Throwable t) - // { - // log.error("Failed to handle failover", t); - // } - // finally - // { - // channel.unlock(); - // } - // - // return ok; - // } + // Needs to be synchronized to prevent issues with occurring concurrently with close() + + //TODO - need to reenable + public synchronized boolean handleReattach(final RemotingConnection backupConnection) + { + if (closed) + { + return true; + } + boolean ok = false; + + // We lock the channel to prevent any packets to be added to the resend + // cache during the failover process + channel.lock(); + + try + { + channel.transferConnection(backupConnection); + + backupConnection.syncIDGeneratorSequence(remotingConnection.getIDGeneratorSequence()); + + remotingConnection = backupConnection; + + Packet request = new ReattachSessionMessage(name, channel.getLastReceivedCommandID()); + + Channel channel1 = backupConnection.getChannel(1, -1, false); + + ReattachSessionResponseMessage response = (ReattachSessionResponseMessage)channel1.sendBlocking(request); + + if (response.isSessionFound()) + { + channel.replayCommands(response.getLastReceivedCommandID(), channel.getID()); + + ok = true; + } + else + { + if (closedSent) + { + // a session re-attach may fail, if the session close was sent before failover started, hit the server, + // processed, then before the response was received back, failover occurred, re-attach was attempted. in + // this case it's ok - we don't want to call any failure listeners and we don't want to halt the rest of + // the failover process. + // + // however if session re-attach fails and the session was not in a call to close, then we DO want to call + // the session listeners so we return false + // + // Also session reattach will fail if the server is restarted - so the session is lost + ok = true; + } + else + { + log.warn(System.identityHashCode(this) + " Session not found on server when attempting to re-attach"); + } + + channel.returnBlocking(); + } + + } + catch (Throwable t) + { + log.error("Failed to handle failover", t); + } + finally + { + channel.unlock(); + } + + return ok; + } + public void workDone() { workDone = true; @@ -861,6 +865,8 @@ boolean ok = false; + log.info("Failover occurring"); + // Need to stop all consumers outside the lock for (ClientConsumerInternal consumer : consumers.values()) { @@ -875,13 +881,21 @@ consumer.clearAtFailover(); } + + log.info("stopped consumers"); // We lock the channel to prevent any packets being sent during the failover process channel.lock(); + + log.info("got lock"); try { + log.info("transferring connection"); + channel.transferConnection(backupConnection); + + log.info("transferred connection"); remotingConnection = backupConnection; @@ -899,10 +913,15 @@ Channel channel1 = backupConnection.getChannel(1, -1, false); + log.info("sending create session"); + CreateSessionResponseMessage response = (CreateSessionResponseMessage)channel1.sendBlocking(request); + log.info("got response from create session"); + if (response.isCreated()) { + log.info("craeted ok"); // Session was created ok // Now we need to recreate the consumers @@ -971,14 +990,18 @@ } ok = true; + + log.info("session created ok"); } else { + log.info("not created ok"); // This means the server we failed onto is not ready to take new sessions - perhaps it hasn't actually // failed over } // We cause any blocking calls to return - since they won't get responses. + log.info("calling returnblocking"); channel.returnBlocking(); } catch (Throwable t) Modified: branches/Branch_Replication_Changes/src/main/org/hornetq/core/client/impl/ClientSessionInternal.java =================================================================== --- branches/Branch_Replication_Changes/src/main/org/hornetq/core/client/impl/ClientSessionInternal.java 2009-09-22 13:57:55 UTC (rev 7978) +++ branches/Branch_Replication_Changes/src/main/org/hornetq/core/client/impl/ClientSessionInternal.java 2009-09-22 15:08:41 UTC (rev 7979) @@ -54,6 +54,8 @@ void handleReceiveContinuation(long consumerID, SessionReceiveContinuationMessage continuation) throws Exception; boolean handleFailover(RemotingConnection backupConnection); + + boolean handleReattach(RemotingConnection backupConnection); RemotingConnection getConnection(); Modified: branches/Branch_Replication_Changes/src/main/org/hornetq/core/client/impl/ConnectionManagerImpl.java =================================================================== --- branches/Branch_Replication_Changes/src/main/org/hornetq/core/client/impl/ConnectionManagerImpl.java 2009-09-22 13:57:55 UTC (rev 7978) +++ branches/Branch_Replication_Changes/src/main/org/hornetq/core/client/impl/ConnectionManagerImpl.java 2009-09-22 15:08:41 UTC (rev 7979) @@ -38,6 +38,7 @@ import org.hornetq.core.remoting.Channel; import org.hornetq.core.remoting.ChannelHandler; import org.hornetq.core.remoting.FailureListener; +import org.hornetq.core.remoting.Interceptor; import org.hornetq.core.remoting.Packet; import org.hornetq.core.remoting.RemotingConnection; import org.hornetq.core.remoting.impl.AbstractBufferHandler; @@ -65,7 +66,6 @@ * * Created 27 Nov 2008 18:46:06 * - * */ public class ConnectionManagerImpl implements ConnectionManager, ConnectionLifeCycleListener { @@ -79,6 +79,7 @@ // Attributes // ----------------------------------------------------------------------------------- + //We hold this reference for GC reasons private final ClientSessionFactory sessionFactory; private final TransportConfiguration connectorConfig; @@ -140,6 +141,10 @@ private PingRunnable pingRunnable; private volatile boolean exitLoop; + + private final List<Interceptor> interceptors; + + private final boolean useReattach; // debug @@ -171,8 +176,10 @@ final long retryInterval, final double retryIntervalMultiplier, final int reconnectAttempts, + final boolean useReattach, final ExecutorService threadPool, - final ScheduledExecutorService scheduledThreadPool) + final ScheduledExecutorService scheduledThreadPool, + final List<Interceptor> interceptors) { this.sessionFactory = sessionFactory; @@ -212,12 +219,16 @@ this.retryIntervalMultiplier = retryIntervalMultiplier; this.reconnectAttempts = reconnectAttempts; + + this.useReattach = useReattach; this.scheduledThreadPool = scheduledThreadPool; this.threadPool = threadPool; this.orderedExecutorFactory = new OrderedExecutorFactory(threadPool); + + this.interceptors = interceptors; } // ConnectionLifeCycleListener implementation -------------------------------------------------- @@ -274,7 +285,7 @@ Channel channel1; synchronized (failoverLock) - { + { connection = getConnectionWithRetry(1, reconnectAttempts); if (connection == null) @@ -287,6 +298,7 @@ throw new HornetQException(HornetQException.NOT_CONNECTED, "Unable to connect to server using configuration " + connectorConfig); + } channel1 = connection.getChannel(1, -1, false); @@ -619,11 +631,11 @@ backupTransportParams = null; - done = reattachSessions(reconnectAttempts == -1 ? -1 : reconnectAttempts + 1); + done = reattachSessions(reconnectAttempts == -1 ? -1 : reconnectAttempts + 1, false); } else if (reconnectAttempts != 0) { - done = reattachSessions(reconnectAttempts); + done = reattachSessions(reconnectAttempts, useReattach); } if (done) @@ -684,7 +696,7 @@ /* * Re-attach sessions all pre-existing sessions to new remoting connections */ - private boolean reattachSessions(final int reconnectAttempts) + private boolean reattachSessions(final int reconnectAttempts, final boolean reattach) { // We re-attach sessions per connection to ensure there is the same mapping of channel id // on live and backup connections @@ -753,8 +765,17 @@ // If all connections got ok, then handle failover for (Map.Entry<ClientSessionInternal, RemotingConnection> entry : sessions.entrySet()) { - boolean b = entry.getKey().handleFailover(entry.getValue()); - + boolean b; + + if (reattach) + { + b = entry.getKey().handleReattach(entry.getValue()); + } + else + { + b = entry.getKey().handleFailover(entry.getValue()); + } + if (!b) { // If a session fails to re-attach we doom the lot, but we make sure we try all sessions and don't exit @@ -956,7 +977,7 @@ return null; } - conn = new RemotingConnectionImpl(tc, callTimeout, null); + conn = new RemotingConnectionImpl(tc, callTimeout, interceptors); conn.addFailureListener(new DelegatingFailureListener(conn.getID())); Modified: branches/Branch_Replication_Changes/src/main/org/hornetq/core/client/impl/DelegatingSession.java =================================================================== --- branches/Branch_Replication_Changes/src/main/org/hornetq/core/client/impl/DelegatingSession.java 2009-09-22 13:57:55 UTC (rev 7978) +++ branches/Branch_Replication_Changes/src/main/org/hornetq/core/client/impl/DelegatingSession.java 2009-09-22 15:08:41 UTC (rev 7979) @@ -340,6 +340,11 @@ { return session.handleFailover(backupConnection); } + + public boolean handleReattach(RemotingConnection backupConnection) + { + return session.handleReattach(backupConnection); + } public void handleReceiveContinuation(long consumerID, SessionReceiveContinuationMessage continuation) throws Exception { Modified: branches/Branch_Replication_Changes/src/main/org/hornetq/core/remoting/impl/RemotingConnectionImpl.java =================================================================== --- branches/Branch_Replication_Changes/src/main/org/hornetq/core/remoting/impl/RemotingConnectionImpl.java 2009-09-22 13:57:55 UTC (rev 7978) +++ branches/Branch_Replication_Changes/src/main/org/hornetq/core/remoting/impl/RemotingConnectionImpl.java 2009-09-22 15:08:41 UTC (rev 7979) @@ -353,36 +353,36 @@ private void doBufferReceived(final Packet packet) { - synchronized (transferLock) - { - if (interceptors != null) + if (interceptors != null) + { + for (final Interceptor interceptor : interceptors) { - for (final Interceptor interceptor : interceptors) + try { - try + boolean callNext = interceptor.intercept(packet, this); + + if (!callNext) { - boolean callNext = interceptor.intercept(packet, this); + // abort - if (!callNext) - { - // abort - - return; - } + return; } - catch (final Throwable e) - { - log.warn("Failure in calling interceptor: " + interceptor, e); - } } + catch (final Throwable e) + { + log.warn("Failure in calling interceptor: " + interceptor, e); + } } - + } + + synchronized (transferLock) + { final Channel channel = channels.get(packet.getChannelID()); if (channel != null) { channel.handlePacket(packet); - } + } } } Modified: branches/Branch_Replication_Changes/src/main/org/hornetq/core/remoting/server/impl/RemotingServiceImpl.java =================================================================== --- branches/Branch_Replication_Changes/src/main/org/hornetq/core/remoting/server/impl/RemotingServiceImpl.java 2009-09-22 13:57:55 UTC (rev 7978) +++ branches/Branch_Replication_Changes/src/main/org/hornetq/core/remoting/server/impl/RemotingServiceImpl.java 2009-09-22 15:08:41 UTC (rev 7979) @@ -22,6 +22,7 @@ import java.util.Map; import java.util.Set; import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.Executor; import java.util.concurrent.ScheduledExecutorService; @@ -71,7 +72,7 @@ private final Set<TransportConfiguration> transportConfigs; - private final List<Interceptor> interceptors = new ArrayList<Interceptor>(); + private final List<Interceptor> interceptors = new CopyOnWriteArrayList<Interceptor>(); private final Set<Acceptor> acceptors = new HashSet<Acceptor>(); Modified: branches/Branch_Replication_Changes/src/main/org/hornetq/core/server/cluster/impl/BridgeImpl.java =================================================================== --- branches/Branch_Replication_Changes/src/main/org/hornetq/core/server/cluster/impl/BridgeImpl.java 2009-09-22 13:57:55 UTC (rev 7978) +++ branches/Branch_Replication_Changes/src/main/org/hornetq/core/server/cluster/impl/BridgeImpl.java 2009-09-22 15:08:41 UTC (rev 7979) @@ -113,7 +113,7 @@ private final double retryIntervalMultiplier; private final int reconnectAttempts; - + private final boolean failoverOnServerShutdown; private final SimpleString idsHeaderName; @@ -131,7 +131,7 @@ private boolean activated; private NotificationService notificationService; - + // Static -------------------------------------------------------- // Constructors -------------------------------------------------- @@ -184,9 +184,9 @@ this.useDuplicateDetection = useDuplicateDetection; this.discoveryAddress = discoveryAddress; - + this.discoveryPort = discoveryPort; - + this.connectorPair = connectorPair; this.retryInterval = retryInterval; @@ -209,34 +209,34 @@ this.flowRecord = flowRecord; - this.activated = activated; + this.activated = activated; } public void setNotificationService(final NotificationService notificationService) { this.notificationService = notificationService; } - + public synchronized void start() throws Exception { if (started) { return; } - + started = true; if (activated) { executor.execute(new CreateObjectsRunnable()); } - + if (notificationService != null) { TypedProperties props = new TypedProperties(); props.putStringProperty(new SimpleString("name"), name); Notification notification = new Notification(nodeUUID.toString(), NotificationType.BRIDGE_STARTED, props); - notificationService.sendNotification(notification ); + notificationService.sendNotification(notification); } } @@ -269,11 +269,11 @@ csf.close(); } } - + executor.execute(new StopRunnable()); - + waitForRunnablesToComplete(); - + if (notificationService != null) { TypedProperties props = new TypedProperties(); @@ -281,7 +281,7 @@ Notification notification = new Notification(nodeUUID.toString(), NotificationType.BRIDGE_STOPPED, props); try { - notificationService.sendNotification(notification ); + notificationService.sendNotification(notification); } catch (Exception e) { @@ -311,7 +311,7 @@ { return queue; } - + public void setQueue(final Queue queue) { this.queue = queue; @@ -360,7 +360,7 @@ if (ref != null) { - ref.getQueue().acknowledge(ref); + ref.getQueue().acknowledge(ref); } } catch (Exception e) @@ -372,12 +372,12 @@ // Consumer implementation --------------------------------------- public HandleStatus handle(final MessageReference ref) throws Exception - { + { if (filter != null && !filter.match(ref.getMessage())) { return HandleStatus.NO_MATCH; } - + if (!active) { return HandleStatus.BUSY; @@ -456,7 +456,7 @@ // Preserve the original address dest = message.getDestination(); } - + producer.send(dest, message); return HandleStatus.HANDLED; @@ -492,7 +492,7 @@ } private void fail() - { + { if (started) { executor.execute(new FailRunnable()); @@ -505,11 +505,11 @@ { return false; } - + try { queue.addConsumer(BridgeImpl.this); - + csf = null; if (discoveryAddress != null) { @@ -517,24 +517,17 @@ } else { - csf = new ClientSessionFactoryImpl(connectorPair.a, - connectorPair.b); + csf = new ClientSessionFactoryImpl(connectorPair.a, connectorPair.b); } - + csf.setFailoverOnServerShutdown(failoverOnServerShutdown); csf.setRetryInterval(retryInterval); csf.setRetryIntervalMultiplier(retryIntervalMultiplier); csf.setReconnectAttempts(reconnectAttempts); - //Session is pre-acknowledge - session = (ClientSessionInternal)csf.createSession(clusterUser, - clusterPassword, - false, - true, - true, - true, - 1); - + // Session is pre-acknowledge + session = (ClientSessionInternal)csf.createSession(clusterUser, clusterPassword, false, true, true, true, 1); + if (session == null) { // This can happen if the bridge is shutdown @@ -559,7 +552,7 @@ // Otherwise it may already exist if server is restarted before it has been deleted on backup String qName = "notif." + nodeUUID.toString() + "." + name.toString(); - + SimpleString notifQueueName = new SimpleString(qName); SimpleString filter = new SimpleString(ManagementHelper.HDR_BINDING_TYPE + "<>" + @@ -584,10 +577,10 @@ flowRecord.getAddress() + "%')"); - //The queue can't be temporary, since if the node with the bridge crashes then is restarted quickly - //it might get deleted on the target when it does connection cleanup - - //When the backup activates the queue might already exist, so we catch this and ignore + // The queue can't be temporary, since if the node with the bridge crashes then is restarted quickly + // it might get deleted on the target when it does connection cleanup + + // When the backup activates the queue might already exist, so we catch this and ignore try { session.createQueue(managementNotificationAddress, notifQueueName, filter, false); @@ -596,7 +589,7 @@ { if (me.getCode() == HornetQException.QUEUE_EXISTS) { - //Ok + // Ok } else { @@ -626,7 +619,7 @@ active = true; queue.deliverAsync(executor); - + return true; } catch (Exception e) @@ -653,8 +646,8 @@ } if (session != null) - { - session.close(); + { + session.close(); } started = false; @@ -689,7 +682,7 @@ { try { - // flowRecord.reset(); + // flowRecord.reset(); } catch (Exception e) { @@ -714,7 +707,7 @@ { log.error("Failed to stop", e); } - + if (!createObjects()) { started = false; Modified: branches/Branch_Replication_Changes/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java =================================================================== --- branches/Branch_Replication_Changes/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java 2009-09-22 13:57:55 UTC (rev 7978) +++ branches/Branch_Replication_Changes/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java 2009-09-22 15:08:41 UTC (rev 7979) @@ -32,6 +32,7 @@ import javax.management.MBeanServer; +import org.hornetq.core.client.impl.ConnectionManager; import org.hornetq.core.config.Configuration; import org.hornetq.core.config.cluster.DivertConfiguration; import org.hornetq.core.config.cluster.QueueConfiguration; @@ -186,6 +187,8 @@ private int managementConnectorID; private static AtomicInteger managementConnectorSequence = new AtomicInteger(0); + + private ConnectionManager replicatingConnectionManager; // Constructors // --------------------------------------------------------------------------------- @@ -387,6 +390,7 @@ initialised = false; uuid = null; nodeID = null; + log.info("HornetQ Server version " + getVersion().getFullVersion() + " stopped"); } @@ -507,7 +511,7 @@ "interoperate properly"); return null; } - + if (!checkActivate()) { //Backup server is not ready to accept connections @@ -633,7 +637,85 @@ // log.info("Backup server is now operational"); // } // } - + +// private boolean setupReplicatingConnection() throws Exception +// { +// String backupConnectorName = configuration.getBackupConnectorName(); +// +// if (backupConnectorName != null) +// { +// TransportConfiguration backupConnector = configuration.getConnectorConfigurations().get(backupConnectorName); +// +// if (backupConnector == null) +// { +// log.warn("connector with name '" + backupConnectorName + "' is not defined in the configuration."); +// } +// else +// { +// replicatingConnectionManager = new ConnectionManagerImpl(null, +// backupConnector, +// null, +// false, +// 1, +// ClientSessionFactoryImpl.DEFAULT_CALL_TIMEOUT, +// ClientSessionFactoryImpl.DEFAULT_CLIENT_FAILURE_CHECK_PERIOD, +// ClientSessionFactoryImpl.DEFAULT_CONNECTION_TTL, +// 0, +// 1.0d, +// 0, +// threadPool, +// scheduledPool); +// +// replicatingConnection = replicatingConnectionManager.getConnection(1); +// +// if (replicatingConnection != null) +// { +// replicatingChannel = replicatingConnection.getChannel(2, -1, false); +// +// replicatingConnection.addFailureListener(new FailureListener() +// { +// public void connectionFailed(HornetQException me) +// { +// replicatingChannel.executeOutstandingDelayedResults(); +// } +// }); +// +// // First time we get channel we send a message down it informing the backup of our node id - +// // backup and live must have the same node id +// +// Packet packet = new ReplicateStartupInfoMessage(uuid, storageManager.getCurrentUniqueID()); +// +// final Future future = new Future(); +// +// replicatingChannel.replicatePacket(packet, 1, new Runnable() +// { +// public void run() +// { +// future.run(); +// } +// }); +// +// // This may take a while especially if the journal is large +// boolean ok = future.await(60000); +// +// if (!ok) +// { +// throw new IllegalStateException("Timed out waiting for response from backup for initialisation"); +// } +// } +// else +// { +// log.warn("Backup server MUST be started before live server. Initialisation will not proceed."); +// +// return false; +// } +// } +// } +// +// return true; +// } + + public HornetQServerControlImpl getHornetQServerControl() { return messagingServerControl; @@ -771,7 +853,7 @@ // -------------------------------------------------------------------------------------- private synchronized boolean checkActivate() throws Exception - { + { if (configuration.isBackup()) { //Handle backup server activation Modified: branches/Branch_Replication_Changes/src/main/org/hornetq/integration/transports/netty/VirtualExecutorService.java =================================================================== --- branches/Branch_Replication_Changes/src/main/org/hornetq/integration/transports/netty/VirtualExecutorService.java 2009-09-22 13:57:55 UTC (rev 7978) +++ branches/Branch_Replication_Changes/src/main/org/hornetq/integration/transports/netty/VirtualExecutorService.java 2009-09-22 15:08:41 UTC (rev 7979) @@ -22,6 +22,7 @@ import java.util.concurrent.RejectedExecutionException; import java.util.concurrent.TimeUnit; +import org.hornetq.core.logging.Logger; import org.jboss.netty.util.internal.MapBackedSet; /** @@ -71,116 +72,155 @@ * @author Trustin Lee (tlee(a)redhat.com) * @version $Rev: 1315 $, $Date: 2009-06-03 03:48:23 -0400 (Wed, 03 Jun 2009) $ */ -public class VirtualExecutorService extends AbstractExecutorService { +public class VirtualExecutorService extends AbstractExecutorService +{ + private static final Logger log = Logger.getLogger(VirtualExecutorService.class); - private final Executor e; - private final ExecutorService s; - final Object startStopLock = new Object(); - volatile boolean shutdown; - Set<Thread> activeThreads = new MapBackedSet<Thread>(new IdentityHashMap<Thread, Boolean>()); + private final Executor e; - /** - * Creates a new instance with the specified parent {@link Executor}. - */ - public VirtualExecutorService(Executor parent) { - if (parent == null) { - throw new NullPointerException("parent"); - } + private final ExecutorService s; - if (parent instanceof ExecutorService) { - e = null; - s = (ExecutorService) parent; - } else { - e = parent; - s = null; - } - } + final Object startStopLock = new Object(); - public boolean isShutdown() { - synchronized (startStopLock) { - return shutdown; - } - } + volatile boolean shutdown; - public boolean isTerminated() { - synchronized (startStopLock) { - return shutdown && activeThreads.isEmpty(); - } - } + Set<Thread> activeThreads = new MapBackedSet<Thread>(new IdentityHashMap<Thread, Boolean>()); - public void shutdown() { - synchronized (startStopLock) { - if (shutdown) { - return; - } - shutdown = true; - } - } + /** + * Creates a new instance with the specified parent {@link Executor}. + */ + public VirtualExecutorService(Executor parent) + { + if (parent == null) + { + throw new NullPointerException("parent"); + } - public List<Runnable> shutdownNow() { - synchronized (startStopLock) { - if (!isTerminated()) { - shutdown(); - for (Thread t: activeThreads) { - t.interrupt(); - } - } - } + if (parent instanceof ExecutorService) + { + e = null; + s = (ExecutorService)parent; + } + else + { + e = parent; + s = null; + } + } - return Collections.emptyList(); - } + public boolean isShutdown() + { + synchronized (startStopLock) + { + return shutdown; + } + } - public boolean awaitTermination(long timeout, TimeUnit unit) - throws InterruptedException { - synchronized (startStopLock) { - while (!isTerminated()) { - startStopLock.wait(TimeUnit.MILLISECONDS.convert(timeout, unit)); + public boolean isTerminated() + { + synchronized (startStopLock) + { + return shutdown && activeThreads.isEmpty(); + } + } + + public void shutdown() + { + synchronized (startStopLock) + { + if (shutdown) + { + return; + } + shutdown = true; + } + } + + public List<Runnable> shutdownNow() + { + synchronized (startStopLock) + { + if (!isTerminated()) + { + shutdown(); + for (Thread t : activeThreads) + { + t.interrupt(); } + } + } - return isTerminated(); - } - } + return Collections.emptyList(); + } - public void execute(Runnable command) { - if (command == null) { - throw new NullPointerException("command"); - } + public boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException + { + synchronized (startStopLock) + { + while (!isTerminated()) + { + startStopLock.wait(TimeUnit.MILLISECONDS.convert(timeout, unit)); + } - if (shutdown) { - throw new RejectedExecutionException(); - } + return isTerminated(); + } + } - if (s != null) { - s.execute(new ChildExecutorRunnable(command)); - } else { - e.execute(new ChildExecutorRunnable(command)); - } - } + public void execute(Runnable command) + { + if (command == null) + { + throw new NullPointerException("command"); + } - private class ChildExecutorRunnable implements Runnable { + if (shutdown) + { + throw new RejectedExecutionException(); + } - private final Runnable runnable; + if (s != null) + { + s.execute(new ChildExecutorRunnable(command)); + } + else + { + e.execute(new ChildExecutorRunnable(command)); + } + } - ChildExecutorRunnable(Runnable runnable) { - this.runnable = runnable; - } + private class ChildExecutorRunnable implements Runnable + { - public void run() { - Thread thread = Thread.currentThread(); - synchronized (startStopLock) { - activeThreads.add(thread); + private final Runnable runnable; + + ChildExecutorRunnable(Runnable runnable) + { + this.runnable = runnable; + } + + public void run() + { + Thread thread = Thread.currentThread(); + synchronized (startStopLock) + { + activeThreads.add(thread); + } + try + { + runnable.run(); + } + finally + { + synchronized (startStopLock) + { + boolean removed = activeThreads.remove(thread); + assert removed; + if (isTerminated()) + { + startStopLock.notifyAll(); + } } - try { - runnable.run(); - } finally { - synchronized (startStopLock) { - boolean removed = activeThreads.remove(thread); - assert removed; - if (isTerminated()) { - startStopLock.notifyAll(); - } - } - } - } - } + } + } + } } \ No newline at end of file Modified: branches/Branch_Replication_Changes/tests/jms-tests/src/org/hornetq/jms/tests/message/MessageHeaderTest.java =================================================================== --- branches/Branch_Replication_Changes/tests/jms-tests/src/org/hornetq/jms/tests/message/MessageHeaderTest.java 2009-09-22 13:57:55 UTC (rev 7978) +++ branches/Branch_Replication_Changes/tests/jms-tests/src/org/hornetq/jms/tests/message/MessageHeaderTest.java 2009-09-22 15:08:41 UTC (rev 7979) @@ -821,6 +821,18 @@ class FakeSession implements ClientSession { + public ClientConsumer createConsumer(SimpleString queueName, boolean browseOnly) throws HornetQException + { + // TODO Auto-generated method stub + return null; + } + + public ClientConsumer createConsumer(String queueName, boolean browseOnly) throws HornetQException + { + // TODO Auto-generated method stub + return null; + } + public void createQueue(String address, String queueName) throws HornetQException { // TODO Auto-generated method stub Modified: branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/integration/clientcrash/DummyInterceptor.java =================================================================== --- branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/integration/clientcrash/DummyInterceptor.java 2009-09-22 13:57:55 UTC (rev 7978) +++ branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/integration/clientcrash/DummyInterceptor.java 2009-09-22 15:08:41 UTC (rev 7979) @@ -62,6 +62,6 @@ } } return true; - } + } } Modified: branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/integration/clientcrash/DummyInterceptorB.java =================================================================== --- branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/integration/clientcrash/DummyInterceptorB.java 2009-09-22 13:57:55 UTC (rev 7978) +++ branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/integration/clientcrash/DummyInterceptorB.java 2009-09-22 15:08:41 UTC (rev 7979) @@ -49,5 +49,4 @@ log.debug("DummyFilter packet = " + packet); return true; } - } Added: branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/integration/cluster/failover/AsynchronousFailoverTest.java =================================================================== --- branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/integration/cluster/failover/AsynchronousFailoverTest.java (rev 0) +++ branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/integration/cluster/failover/AsynchronousFailoverTest.java 2009-09-22 15:08:41 UTC (rev 7979) @@ -0,0 +1,415 @@ +/* + * Copyright 2009 Red Hat, Inc. + * Red Hat licenses this file to you under the Apache License, version + * 2.0 (the "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * http://www.apache.org/licenses/LICENSE-2.0 + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or + * implied. See the License for the specific language governing + * permissions and limitations under the License. + */ + +package org.hornetq.tests.integration.cluster.failover; + +import java.util.HashMap; +import java.util.Map; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; + +import org.hornetq.core.client.ClientConsumer; +import org.hornetq.core.client.ClientMessage; +import org.hornetq.core.client.ClientProducer; +import org.hornetq.core.client.ClientSession; +import org.hornetq.core.client.impl.ClientSessionFactoryInternal; +import org.hornetq.core.client.impl.ClientSessionInternal; +import org.hornetq.core.config.TransportConfiguration; +import org.hornetq.core.exception.HornetQException; +import org.hornetq.core.logging.Logger; +import org.hornetq.core.remoting.FailureListener; +import org.hornetq.core.remoting.RemotingConnection; +import org.hornetq.core.remoting.impl.invm.TransportConstants; + +/** + * A MultiThreadFailoverTest + * + * Test Failover where failure is prompted by another thread + * + * @author Tim Fox + * + * + */ +public class AsynchronousFailoverTest extends FailoverTestBase +{ + private static final Logger log = Logger.getLogger(AsynchronousFailoverTest.class); + + private volatile MyListener listener; + + private volatile ClientSessionFactoryInternal sf; + + class MyListener implements FailureListener + { + CountDownLatch latch = new CountDownLatch(1); + + public void connectionFailed(HornetQException me) + { + latch.countDown(); + } + } + + public void testNonTransactional() throws Exception + { + runTest(new TestRunner() + { + public void run() + { + try + { + doTestNonTransactional(this); + } + catch (Exception e) + { + log.error("Test failed", e); + } + } + }); + } + + public void testTransactional() throws Exception + { + runTest(new TestRunner() + { + public void run() + { + try + { + doTestTransactional(this); + } + catch (Exception e) + { + log.error("Test failed", e); + } + } + }); + } + + abstract class TestRunner implements Runnable + { + volatile boolean failed; + + boolean isFailed() + { + return failed; + } + + void setFailed() + { + failed = true; + } + } + + private void runTest(final TestRunner runnable) throws Exception + { + final int numIts = 10; + + for (int i = 0; i < numIts; i++) + { + log.info("Iteration " + i); + + sf = getSessionFactory(); + + sf.setBlockOnNonPersistentSend(true); + sf.setBlockOnPersistentSend(true); + + ClientSession createSession = sf.createSession(true, true); + + createSession.createQueue(ADDRESS, ADDRESS, null, true); + + RemotingConnection conn = ((ClientSessionInternal)createSession).getConnection(); + + Thread t = new Thread(runnable); + + t.start(); + + long randomDelay = (long)(2000 * Math.random()); + + log.info("Sleeping " + randomDelay); + + Thread.sleep(randomDelay); + + log.info("Failing asynchronously"); + + MyListener listener = this.listener; + + // Simulate failure on connection + conn.fail(new HornetQException(HornetQException.NOT_CONNECTED)); + + if (listener != null) + { + boolean ok = listener.latch.await(10000, TimeUnit.MILLISECONDS); + + assertTrue(ok); + } + + runnable.setFailed(); + + log.info("Fail complete"); + + t.join(); + + createSession.close(); + + assertEquals(0, sf.numSessions()); + + assertEquals(0, sf.numConnections()); + + if (i != numIts - 1) + { + tearDown(); + setUp(); + } + } + } + + private void doTestNonTransactional(final TestRunner runner) throws Exception + { + while (!runner.isFailed()) + { + log.info("looping"); + + ClientSession session = sf.createSession(true, true, 0); + + MyListener listener = new MyListener(); + + session.addFailureListener(listener); + + this.listener = listener; + + ClientProducer producer = session.createProducer(ADDRESS); + + final int numMessages = 1000; + + for (int i = 0; i < numMessages; i++) + { + ClientMessage message = session.createClientMessage(true); + + message.getBody().writeString("message" + i); + + message.putIntProperty("counter", i); + + boolean retry = false; + do + { + try + { + producer.send(message); + + retry = false; + } + catch (HornetQException e) + { + assertEquals(e.getCode(), HornetQException.UNBLOCKED); + + retry = true; + } + } + while (retry); + } + + ClientConsumer consumer = session.createConsumer(ADDRESS); + + session.start(); + + int lastCount = -1; + while (true) + { + ClientMessage message = consumer.receive(500); + + if (message == null) + { + break; + } + + // There may be some missing or duplicate messages - but the order should be correct + + int count = (Integer)message.getProperty("counter"); + + assertTrue("count:" + count + " last count:" + lastCount, count >= lastCount); + + lastCount = count; + + message.acknowledge(); + } + + session.close(); + + this.listener = null; + } + } + + private void doTestTransactional(final TestRunner runner) throws Exception + { + while (!runner.isFailed()) + { + ClientSession session = sf.createSession(false, false); + + MyListener listener = new MyListener(); + + session.addFailureListener(listener); + + this.listener = listener; + + ClientProducer producer = session.createProducer(ADDRESS); + + final int numMessages = 1000; + + for (int i = 0; i < numMessages; i++) + { + ClientMessage message = session.createClientMessage(true); + + message.getBody().writeString("message" + i); + + message.putIntProperty("counter", i); + + boolean retry = false; + do + { + try + { + producer.send(message); + } + catch (HornetQException e) + { + assertEquals(e.getCode(), HornetQException.UNBLOCKED); + + retry = true; + } + } + while (retry); + } + + boolean retry = false; + while (retry) + { + try + { + session.commit(); + + retry = false; + } + catch (HornetQException e) + { + if (e.getCode() == HornetQException.TRANSACTION_ROLLED_BACK) + { + // OK + session.close(); + + continue; + } + else if (e.getCode() == HornetQException.UNBLOCKED) + { + retry = true; + } + else + { + throw e; + } + } + } + + ClientConsumer consumer = session.createConsumer(ADDRESS); + + session.start(); + + int lastCount = -1; + while (true) + { + ClientMessage message = consumer.receive(500); + + if (message == null) + { + break; + } + + // There may be some missing or duplicate messages - but the order should be correct + + int count = (Integer)message.getProperty("counter"); + + assertTrue("count:" + count + " last count:" + lastCount, count >= lastCount); + + lastCount = count; + + message.acknowledge(); + } + + retry = false; + while (retry) + { + try + { + session.commit(); + + retry = false; + } + catch (HornetQException e) + { + if (e.getCode() == HornetQException.TRANSACTION_ROLLED_BACK) + { + // OK + session.close(); + + continue; + } + else if (e.getCode() == HornetQException.UNBLOCKED) + { + retry = true; + } + else + { + throw e; + } + } + } + + session.close(); + + this.listener = null; + } + } + + @Override + protected TransportConfiguration getAcceptorTransportConfiguration(boolean live) + { + if (live) + { + return new TransportConfiguration("org.hornetq.core.remoting.impl.invm.InVMAcceptorFactory"); + } + else + { + Map<String, Object> server1Params = new HashMap<String, Object>(); + + server1Params.put(TransportConstants.SERVER_ID_PROP_NAME, 1); + + return new TransportConfiguration("org.hornetq.core.remoting.impl.invm.InVMAcceptorFactory", server1Params); + } + } + + @Override + protected TransportConfiguration getConnectorTransportConfiguration(final boolean live) + { + if (live) + { + return new TransportConfiguration("org.hornetq.core.remoting.impl.invm.InVMConnectorFactory"); + } + else + { + Map<String, Object> server1Params = new HashMap<String, Object>(); + + server1Params.put(TransportConstants.SERVER_ID_PROP_NAME, 1); + + return new TransportConfiguration("org.hornetq.core.remoting.impl.invm.InVMConnectorFactory", server1Params); + } + } + +} Added: branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/integration/cluster/failover/DelayInterceptor2.java =================================================================== --- branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/integration/cluster/failover/DelayInterceptor2.java (rev 0) +++ branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/integration/cluster/failover/DelayInterceptor2.java 2009-09-22 15:08:41 UTC (rev 7979) @@ -0,0 +1,47 @@ +/* + * Copyright 2009 Red Hat, Inc. + * Red Hat licenses this file to you under the Apache License, version + * 2.0 (the "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * http://www.apache.org/licenses/LICENSE-2.0 + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or + * implied. See the License for the specific language governing + * permissions and limitations under the License. + */ + +package org.hornetq.tests.integration.cluster.failover; + +import org.hornetq.core.exception.HornetQException; +import org.hornetq.core.logging.Logger; +import org.hornetq.core.remoting.Interceptor; +import org.hornetq.core.remoting.Packet; +import org.hornetq.core.remoting.RemotingConnection; + +/** + * A DelayInterceptor2 + * + * @author <a href="mailto:tim.fox@jboss.com">Tim Fox</a> + * + * + */ +public class DelayInterceptor2 implements Interceptor +{ + private static final Logger log = Logger.getLogger(DelayInterceptor2.class); + + public boolean intercept(Packet packet, RemotingConnection connection) throws HornetQException + { + try + { + Thread.sleep(2000); + } + catch (InterruptedException e) + { + } + + log.info("proceeding"); + + return true; + } +} Added: branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/integration/cluster/failover/DelayInterceptor3.java =================================================================== --- branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/integration/cluster/failover/DelayInterceptor3.java (rev 0) +++ branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/integration/cluster/failover/DelayInterceptor3.java 2009-09-22 15:08:41 UTC (rev 7979) @@ -0,0 +1,50 @@ +/* + * Copyright 2009 Red Hat, Inc. + * Red Hat licenses this file to you under the Apache License, version + * 2.0 (the "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * http://www.apache.org/licenses/LICENSE-2.0 + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or + * implied. See the License for the specific language governing + * permissions and limitations under the License. + */ + +package org.hornetq.tests.integration.cluster.failover; + +import org.hornetq.core.exception.HornetQException; +import org.hornetq.core.logging.Logger; +import org.hornetq.core.remoting.Interceptor; +import org.hornetq.core.remoting.Packet; +import org.hornetq.core.remoting.RemotingConnection; +import org.hornetq.core.remoting.impl.wireformat.PacketImpl; + +/** + * A DelayInterceptor3 + * + * @author <a href="mailto:tim.fox@jboss.com">Tim Fox</a> + * + * + */ +public class DelayInterceptor3 implements Interceptor +{ + private static final Logger log = Logger.getLogger(DelayInterceptor3.class); + + public boolean intercept(Packet packet, RemotingConnection connection) throws HornetQException + { + if (packet.getType() == PacketImpl.SESS_COMMIT) + { + log.info("got sess commit, delaying"); + try + { + Thread.sleep(2000); + } + catch (Exception e) + { + } + } + + return false; + } +} Deleted: branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/integration/cluster/failover/FailoverNoSessionsFailoverTest.java =================================================================== --- branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/integration/cluster/failover/FailoverNoSessionsFailoverTest.java 2009-09-22 13:57:55 UTC (rev 7978) +++ branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/integration/cluster/failover/FailoverNoSessionsFailoverTest.java 2009-09-22 15:08:41 UTC (rev 7979) @@ -1,228 +0,0 @@ -/* - * Copyright 2009 Red Hat, Inc. - * Red Hat licenses this file to you under the Apache License, version - * 2.0 (the "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * http://www.apache.org/licenses/LICENSE-2.0 - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or - * implied. See the License for the specific language governing - * permissions and limitations under the License. - */ - -package org.hornetq.tests.integration.cluster.failover; - -import java.util.HashMap; -import java.util.Map; - -import org.hornetq.core.client.ClientConsumer; -import org.hornetq.core.client.ClientMessage; -import org.hornetq.core.client.ClientProducer; -import org.hornetq.core.client.ClientSession; -import org.hornetq.core.client.impl.ClientSessionFactoryImpl; -import org.hornetq.core.client.impl.ClientSessionFactoryInternal; -import org.hornetq.core.config.Configuration; -import org.hornetq.core.config.TransportConfiguration; -import org.hornetq.core.config.impl.ConfigurationImpl; -import org.hornetq.core.logging.Logger; -import org.hornetq.core.remoting.impl.invm.InVMRegistry; -import org.hornetq.core.remoting.impl.invm.TransportConstants; -import org.hornetq.core.server.HornetQ; -import org.hornetq.core.server.HornetQServer; -import org.hornetq.jms.client.HornetQTextMessage; -import org.hornetq.tests.util.UnitTestCase; -import org.hornetq.utils.SimpleString; - -/** - * Test situation where none of the sessions failover, but a new session is created - * - * A FailoverNoSessionsFailoverTest - * - * @author <a href="mailto:tim.fox@jboss.com">Tim Fox</a> - * - * Created 6 Nov 2008 17:39:23 - * - * - */ -public class FailoverNoSessionsFailoverTest extends UnitTestCase -{ - private static final Logger log = Logger.getLogger(FailoverNoSessionsFailoverTest.class); - - // Constants ----------------------------------------------------- - - // Attributes ---------------------------------------------------- - - private static final SimpleString ADDRESS = new SimpleString("FailoverTestAddress"); - - private HornetQServer liveService; - - private HornetQServer backupService; - - private Map<String, Object> backupParams = new HashMap<String, Object>(); - - // Static -------------------------------------------------------- - - // Constructors -------------------------------------------------- - - // Public -------------------------------------------------------- - - public void testNoFailoverAndCreateNewSession() throws Exception - { - ClientSessionFactoryInternal sf1 = new ClientSessionFactoryImpl(new TransportConfiguration("org.hornetq.core.remoting.impl.invm.InVMConnectorFactory"), - new TransportConfiguration("org.hornetq.core.remoting.impl.invm.InVMConnectorFactory", - backupParams)); - - sf1.setProducerWindowSize(32 * 1024); - - ClientSession session1 = sf1.createSession(false, true, true); - - session1.createQueue(ADDRESS, ADDRESS, null, false); - - ClientProducer producer = session1.createProducer(ADDRESS); - - final int numMessages = 1000; - - for (int i = 0; i < numMessages; i++) - { - ClientMessage message = session1.createClientMessage(HornetQTextMessage.TYPE, - false, - 0, - System.currentTimeMillis(), - (byte)1); - message.putIntProperty(new SimpleString("count"), i); - message.getBody().writeString("aardvarks"); - producer.send(message); - } - - ClientConsumer consumer1 = session1.createConsumer(ADDRESS); - - session1.start(); - - for (int i = 0; i < numMessages; i++) - { - ClientMessage message = consumer1.receive(1000); - - assertNotNull(message); - - assertEquals("aardvarks", message.getBody().readString()); - - assertEquals(i, message.getProperty(new SimpleString("count"))); - - message.acknowledge(); - } - - ClientMessage message = consumer1.receive(1000); - - assertNull(message); - - session1.close(); - - //Now create another sesion direct on the backup - - ClientSessionFactoryInternal sf2 = new ClientSessionFactoryImpl(new TransportConfiguration("org.hornetq.core.remoting.impl.invm.InVMConnectorFactory", - backupParams)); - - sf2.setProducerWindowSize(32 * 1024); - - ClientSession session2 = sf2.createSession(false, true, true); - - ClientProducer producer2 = session2.createProducer(ADDRESS); - - for (int i = 0; i < numMessages; i++) - { - ClientMessage message2 = session2.createClientMessage(HornetQTextMessage.TYPE, - false, - 0, - System.currentTimeMillis(), - (byte)1); - message2.putIntProperty(new SimpleString("count"), i); - message2.getBody().writeString("aardvarks"); - producer2.send(message2); - } - - ClientConsumer consumer2 = session2.createConsumer(ADDRESS); - - session2.start(); - - try - { - for (int i = 0; i < numMessages; i++) - { - ClientMessage message2 = consumer2.receive(1000); - - assertNotNull(message2); - - assertEquals("aardvarks", message2.getBody().readString()); - - assertEquals(i, message2.getProperty(new SimpleString("count"))); - - message2.acknowledge(); - } - - ClientMessage message2 = consumer2.receive(1000); - - assertNull(message2); - } - finally - { - session2.close(); - } - } - - // Package protected --------------------------------------------- - - // Protected ----------------------------------------------------- - - @Override - protected void setUp() throws Exception - { - super.setUp(); - - Configuration backupConf = new ConfigurationImpl(); - backupConf.setSecurityEnabled(false); - backupParams.put(TransportConstants.SERVER_ID_PROP_NAME, 1); - backupConf.getAcceptorConfigurations() - .add(new TransportConfiguration("org.hornetq.core.remoting.impl.invm.InVMAcceptorFactory", - backupParams)); - backupConf.setBackup(true); - backupService = HornetQ.newHornetQServer(backupConf, false); - backupService.start(); - - Configuration liveConf = new ConfigurationImpl(); - liveConf.setSecurityEnabled(false); - liveConf.getAcceptorConfigurations() - .add(new TransportConfiguration("org.hornetq.core.remoting.impl.invm.InVMAcceptorFactory")); - Map<String, TransportConfiguration> connectors = new HashMap<String, TransportConfiguration>(); - TransportConfiguration backupTC = new TransportConfiguration("org.hornetq.core.remoting.impl.invm.InVMConnectorFactory", - backupParams, "backup-connector"); - connectors.put(backupTC.getName(), backupTC); - liveConf.setConnectorConfigurations(connectors); - liveConf.setBackupConnectorName(backupTC.getName()); - liveService = HornetQ.newHornetQServer(liveConf, false); - liveService.start(); - } - - @Override - protected void tearDown() throws Exception - { - backupService.stop(); - - liveService.stop(); - - assertEquals(0, InVMRegistry.instance.size()); - - backupService = null; - - liveService = null; - - backupParams = null; - - super.tearDown(); - } - - // Private ------------------------------------------------------- - - // Inner classes ------------------------------------------------- -} - Modified: branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/integration/cluster/failover/FailoverTest.java =================================================================== --- branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/integration/cluster/failover/FailoverTest.java 2009-09-22 13:57:55 UTC (rev 7978) +++ branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/integration/cluster/failover/FailoverTest.java 2009-09-22 15:08:41 UTC (rev 7979) @@ -32,21 +32,16 @@ import org.hornetq.core.client.impl.ClientSessionFactoryImpl; import org.hornetq.core.client.impl.ClientSessionFactoryInternal; import org.hornetq.core.client.impl.ClientSessionInternal; -import org.hornetq.core.config.Configuration; import org.hornetq.core.config.TransportConfiguration; import org.hornetq.core.exception.HornetQException; import org.hornetq.core.logging.Logger; +import org.hornetq.core.message.impl.MessageImpl; import org.hornetq.core.remoting.FailureListener; import org.hornetq.core.remoting.Interceptor; -import org.hornetq.core.remoting.Packet; import org.hornetq.core.remoting.RemotingConnection; -import org.hornetq.core.remoting.impl.invm.InVMRegistry; import org.hornetq.core.remoting.impl.invm.TransportConstants; -import org.hornetq.core.remoting.impl.wireformat.PacketImpl; -import org.hornetq.core.server.HornetQServer; import org.hornetq.core.transaction.impl.XidImpl; import org.hornetq.jms.client.HornetQTextMessage; -import org.hornetq.tests.util.ServiceTestBase; import org.hornetq.utils.SimpleString; /** @@ -69,7 +64,7 @@ * @author <a href="mailto:tim.fox@jboss.com">Tim Fox</a> * */ -public class FailoverTest extends ServiceTestBase +public class FailoverTest extends FailoverTestBase { private static final Logger log = Logger.getLogger(FailoverTest.class); @@ -77,14 +72,6 @@ // Attributes ---------------------------------------------------- - private static final SimpleString ADDRESS = new SimpleString("FailoverTestAddress"); - - private HornetQServer server0Service; - - private HornetQServer server1Service; - - private Map<String, Object> server1Params = new HashMap<String, Object>(); - // Static -------------------------------------------------------- // Constructors -------------------------------------------------- @@ -93,9 +80,7 @@ public void testNonTransacted() throws Exception { - ClientSessionFactoryInternal sf = new ClientSessionFactoryImpl(new TransportConfiguration("org.hornetq.core.remoting.impl.invm.InVMConnectorFactory"), - new TransportConfiguration("org.hornetq.core.remoting.impl.invm.InVMConnectorFactory", - server1Params)); + ClientSessionFactoryInternal sf = getSessionFactory(); sf.setBlockOnNonPersistentSend(true); sf.setBlockOnPersistentSend(true); @@ -173,9 +158,7 @@ public void testTransactedMessagesSentSoRollback() throws Exception { - ClientSessionFactoryInternal sf = new ClientSessionFactoryImpl(new TransportConfiguration("org.hornetq.core.remoting.impl.invm.InVMConnectorFactory"), - new TransportConfiguration("org.hornetq.core.remoting.impl.invm.InVMConnectorFactory", - server1Params)); + ClientSessionFactoryInternal sf = getSessionFactory(); sf.setBlockOnNonPersistentSend(true); sf.setBlockOnPersistentSend(true); @@ -250,9 +233,7 @@ public void testTransactedMessagesNotSentSoNoRollback() throws Exception { - ClientSessionFactoryInternal sf = new ClientSessionFactoryImpl(new TransportConfiguration("org.hornetq.core.remoting.impl.invm.InVMConnectorFactory"), - new TransportConfiguration("org.hornetq.core.remoting.impl.invm.InVMConnectorFactory", - server1Params)); + ClientSessionFactoryInternal sf = getSessionFactory(); sf.setBlockOnNonPersistentSend(true); sf.setBlockOnPersistentSend(true); @@ -338,9 +319,7 @@ public void testTransactedMessagesConsumedSoRollback() throws Exception { - ClientSessionFactoryInternal sf = new ClientSessionFactoryImpl(new TransportConfiguration("org.hornetq.core.remoting.impl.invm.InVMConnectorFactory"), - new TransportConfiguration("org.hornetq.core.remoting.impl.invm.InVMConnectorFactory", - server1Params)); + ClientSessionFactoryInternal sf = getSessionFactory(); sf.setBlockOnNonPersistentSend(true); sf.setBlockOnPersistentSend(true); @@ -430,9 +409,7 @@ public void testTransactedMessagesNotConsumedSoNoRollback() throws Exception { - ClientSessionFactoryInternal sf = new ClientSessionFactoryImpl(new TransportConfiguration("org.hornetq.core.remoting.impl.invm.InVMConnectorFactory"), - new TransportConfiguration("org.hornetq.core.remoting.impl.invm.InVMConnectorFactory", - server1Params)); + ClientSessionFactoryInternal sf = getSessionFactory(); sf.setBlockOnNonPersistentSend(true); sf.setBlockOnPersistentSend(true); @@ -538,9 +515,7 @@ public void testXAMessagesSentSoRollbackOnEnd() throws Exception { - ClientSessionFactoryInternal sf = new ClientSessionFactoryImpl(new TransportConfiguration("org.hornetq.core.remoting.impl.invm.InVMConnectorFactory"), - new TransportConfiguration("org.hornetq.core.remoting.impl.invm.InVMConnectorFactory", - server1Params)); + ClientSessionFactoryInternal sf = getSessionFactory(); sf.setBlockOnNonPersistentSend(true); sf.setBlockOnPersistentSend(true); @@ -619,9 +594,7 @@ public void testXAMessagesSentSoRollbackOnPrepare() throws Exception { - ClientSessionFactoryInternal sf = new ClientSessionFactoryImpl(new TransportConfiguration("org.hornetq.core.remoting.impl.invm.InVMConnectorFactory"), - new TransportConfiguration("org.hornetq.core.remoting.impl.invm.InVMConnectorFactory", - server1Params)); + ClientSessionFactoryInternal sf = getSessionFactory(); sf.setBlockOnNonPersistentSend(true); sf.setBlockOnPersistentSend(true); @@ -703,9 +676,7 @@ // This might happen if 1PC optimisation kicks in public void testXAMessagesSentSoRollbackOnCommit() throws Exception { - ClientSessionFactoryInternal sf = new ClientSessionFactoryImpl(new TransportConfiguration("org.hornetq.core.remoting.impl.invm.InVMConnectorFactory"), - new TransportConfiguration("org.hornetq.core.remoting.impl.invm.InVMConnectorFactory", - server1Params)); + ClientSessionFactoryInternal sf = getSessionFactory(); sf.setBlockOnNonPersistentSend(true); sf.setBlockOnPersistentSend(true); @@ -788,9 +759,7 @@ public void testXAMessagesNotSentSoNoRollbackOnCommit() throws Exception { - ClientSessionFactoryInternal sf = new ClientSessionFactoryImpl(new TransportConfiguration("org.hornetq.core.remoting.impl.invm.InVMConnectorFactory"), - new TransportConfiguration("org.hornetq.core.remoting.impl.invm.InVMConnectorFactory", - server1Params)); + ClientSessionFactoryInternal sf = getSessionFactory(); sf.setBlockOnNonPersistentSend(true); sf.setBlockOnPersistentSend(true); @@ -888,9 +857,7 @@ public void testXAMessagesConsumedSoRollbackOnEnd() throws Exception { - ClientSessionFactoryInternal sf = new ClientSessionFactoryImpl(new TransportConfiguration("org.hornetq.core.remoting.impl.invm.InVMConnectorFactory"), - new TransportConfiguration("org.hornetq.core.remoting.impl.invm.InVMConnectorFactory", - server1Params)); + ClientSessionFactoryInternal sf = getSessionFactory(); sf.setBlockOnNonPersistentSend(true); sf.setBlockOnPersistentSend(true); @@ -984,9 +951,7 @@ public void testXAMessagesConsumedSoRollbackOnPrepare() throws Exception { - ClientSessionFactoryInternal sf = new ClientSessionFactoryImpl(new TransportConfiguration("org.hornetq.core.remoting.impl.invm.InVMConnectorFactory"), - new TransportConfiguration("org.hornetq.core.remoting.impl.invm.InVMConnectorFactory", - server1Params)); + ClientSessionFactoryInternal sf = getSessionFactory(); sf.setBlockOnNonPersistentSend(true); sf.setBlockOnPersistentSend(true); @@ -1083,9 +1048,7 @@ // 1PC optimisation public void testXAMessagesConsumedSoRollbackOnCommit() throws Exception { - ClientSessionFactoryInternal sf = new ClientSessionFactoryImpl(new TransportConfiguration("org.hornetq.core.remoting.impl.invm.InVMConnectorFactory"), - new TransportConfiguration("org.hornetq.core.remoting.impl.invm.InVMConnectorFactory", - server1Params)); + ClientSessionFactoryInternal sf = getSessionFactory(); sf.setBlockOnNonPersistentSend(true); sf.setBlockOnPersistentSend(true); @@ -1183,7 +1146,7 @@ public void testCreateNewFactoryAfterFailover() throws Exception { - ClientSessionFactoryInternal sf = new ClientSessionFactoryImpl(new TransportConfiguration("org.hornetq.core.remoting.impl.invm.InVMConnectorFactory")); + ClientSessionFactoryInternal sf = getSessionFactory(); ClientSession session = sendAndConsume(sf); @@ -1210,10 +1173,11 @@ assertTrue(ok); + log.info("closing session"); session.close(); + log.info("closed session"); - sf = new ClientSessionFactoryImpl(new TransportConfiguration("org.hornetq.core.remoting.impl.invm.InVMConnectorFactory", - server1Params)); + sf = new ClientSessionFactoryImpl(getConnectorTransportConfiguration(false)); session = sendAndConsume(sf); @@ -1226,9 +1190,7 @@ public void testFailoverMultipleSessionsWithConsumers() throws Exception { - ClientSessionFactoryInternal sf = new ClientSessionFactoryImpl(new TransportConfiguration("org.hornetq.core.remoting.impl.invm.InVMConnectorFactory"), - new TransportConfiguration("org.hornetq.core.remoting.impl.invm.InVMConnectorFactory", - server1Params)); + ClientSessionFactoryInternal sf = getSessionFactory(); sf.setBlockOnNonPersistentSend(true); sf.setBlockOnPersistentSend(true); @@ -1339,9 +1301,7 @@ public void testFailoverFailMultipleUnderlyingConnections() throws Exception { - ClientSessionFactoryInternal sf = new ClientSessionFactoryImpl(new TransportConfiguration("org.hornetq.core.remoting.impl.invm.InVMConnectorFactory"), - new TransportConfiguration("org.hornetq.core.remoting.impl.invm.InVMConnectorFactory", - server1Params)); + ClientSessionFactoryInternal sf = getSessionFactory(); sf.setBlockOnNonPersistentSend(true); sf.setBlockOnPersistentSend(true); @@ -1450,107 +1410,12 @@ assertEquals(0, sf.numConnections()); } - // Package protected --------------------------------------------- - - // Protected ----------------------------------------------------- - - protected void setUp() throws Exception - { - super.setUp(); - - server1Params.put(TransportConstants.SERVER_ID_PROP_NAME, 1); - Configuration config1 = super.createDefaultConfig(); - config1.getAcceptorConfigurations().clear(); - config1.getAcceptorConfigurations() - .add(new TransportConfiguration("org.hornetq.core.remoting.impl.invm.InVMAcceptorFactory", server1Params)); - config1.setSecurityEnabled(false); - config1.setSharedStore(true); - config1.setBackup(true); - server1Service = super.createServer(true, config1); - - Configuration config0 = super.createDefaultConfig(); - config0.getAcceptorConfigurations().clear(); - config0.getAcceptorConfigurations() - .add(new TransportConfiguration("org.hornetq.core.remoting.impl.invm.InVMAcceptorFactory")); - config0.setSecurityEnabled(false); - config0.setSharedStore(true); - server0Service = super.createServer(true, config0); - - server1Service.start(); - server0Service.start(); - } - - protected void tearDown() throws Exception - { - server1Service.stop(); - - server0Service.stop(); - - assertEquals(0, InVMRegistry.instance.size()); - - server1Service = null; - - server0Service = null; - - server1Params = null; - - super.tearDown(); - } - - // Private ------------------------------------------------------- - - private ClientSession sendAndConsume(final ClientSessionFactory sf) throws Exception - { - ClientSession session = sf.createSession(false, true, true); - - session.createQueue(ADDRESS, ADDRESS, null, false); - - ClientProducer producer = session.createProducer(ADDRESS); - - final int numMessages = 1000; - - for (int i = 0; i < numMessages; i++) - { - ClientMessage message = session.createClientMessage(HornetQTextMessage.TYPE, - false, - 0, - System.currentTimeMillis(), - (byte)1); - message.putIntProperty(new SimpleString("count"), i); - message.getBody().writeString("aardvarks"); - producer.send(message); - } - - ClientConsumer consumer = session.createConsumer(ADDRESS); - - session.start(); - - for (int i = 0; i < numMessages; i++) - { - ClientMessage message2 = consumer.receive(); - - assertEquals("aardvarks", message2.getBody().readString()); - - assertEquals(i, message2.getProperty(new SimpleString("count"))); - - message2.acknowledge(); - } - - ClientMessage message3 = consumer.receive(250); - - assertNull(message3); - - return session; - } - /* * Browser will get reset to beginning after failover */ public void testFailWithBrowser() throws Exception { - ClientSessionFactoryInternal sf = new ClientSessionFactoryImpl(new TransportConfiguration("org.hornetq.core.remoting.impl.invm.InVMConnectorFactory"), - new TransportConfiguration("org.hornetq.core.remoting.impl.invm.InVMConnectorFactory", - server1Params)); + ClientSessionFactoryInternal sf = this.getSessionFactory(); sf.setBlockOnNonPersistentSend(true); sf.setBlockOnPersistentSend(true); @@ -1643,9 +1508,7 @@ public void testFailThenReceiveMoreMessagesAfterFailover() throws Exception { - ClientSessionFactoryInternal sf = new ClientSessionFactoryImpl(new TransportConfiguration("org.hornetq.core.remoting.impl.invm.InVMConnectorFactory"), - new TransportConfiguration("org.hornetq.core.remoting.impl.invm.InVMConnectorFactory", - server1Params)); + ClientSessionFactoryInternal sf = getSessionFactory(); sf.setBlockOnNonPersistentSend(true); sf.setBlockOnPersistentSend(true); @@ -1736,9 +1599,7 @@ public void testFailThenReceiveMoreMessagesAfterFailover2() throws Exception { - ClientSessionFactoryInternal sf = new ClientSessionFactoryImpl(new TransportConfiguration("org.hornetq.core.remoting.impl.invm.InVMConnectorFactory"), - new TransportConfiguration("org.hornetq.core.remoting.impl.invm.InVMConnectorFactory", - server1Params)); + ClientSessionFactoryInternal sf = getSessionFactory(); sf.setBlockOnNonPersistentSend(true); sf.setBlockOnPersistentSend(true); @@ -1788,7 +1649,7 @@ assertEquals("message" + i, message.getBody().readString()); assertEquals(i, message.getProperty("counter")); - + message.acknowledge(); } @@ -1837,21 +1698,15 @@ assertEquals(0, sf.numConnections()); } - + public void testForceBlockingReturn() throws Exception - { - server0Service.stop(); - - //Add an interceptor to delay the send method so we can get time to cause failover before it returns - - server0Service.getConfiguration().getInterceptorClassNames().add(DelayInterceptor.class.getCanonicalName()); - - ClientSessionFactoryInternal sf = new ClientSessionFactoryImpl(new TransportConfiguration("org.hornetq.core.remoting.impl.invm.InVMConnectorFactory"), - new TransportConfiguration("org.hornetq.core.remoting.impl.invm.InVMConnectorFactory", - server1Params)); - - server0Service.start(); + { + ClientSessionFactoryInternal sf = this.getSessionFactory(); + // Add an interceptor to delay the send method so we can get time to cause failover before it returns + + server0Service.getRemotingService().addInterceptor(new DelayInterceptor()); + sf.setBlockOnNonPersistentSend(true); sf.setBlockOnPersistentSend(true); sf.setBlockOnAcknowledge(true); @@ -1873,7 +1728,7 @@ session.addFailureListener(new MyListener()); final ClientProducer producer = session.createProducer(ADDRESS); - + class Sender extends Thread { public void run() @@ -1891,16 +1746,16 @@ this.e = e; } } - + volatile HornetQException e; } - + Sender sender = new Sender(); - + sender.start(); - + Thread.sleep(500); - + RemotingConnection conn = ((ClientSessionInternal)session).getConnection(); // Simulate failure on connection @@ -1911,22 +1766,427 @@ boolean ok = latch.await(1000, TimeUnit.MILLISECONDS); assertTrue(ok); - + sender.join(); - + assertNotNull(sender.e); - + assertEquals(sender.e.getCode(), HornetQException.UNBLOCKED); - + session.close(); assertEquals(0, sf.numSessions()); assertEquals(0, sf.numConnections()); - - } - - + + public void testCommitOccurredUnblockedAndResendNoDuplicates() throws Exception + { + final ClientSessionFactoryInternal sf = this.getSessionFactory(); + + sf.setBlockOnNonPersistentSend(true); + sf.setBlockOnPersistentSend(true); + sf.setBlockOnAcknowledge(true); + + log.info("creating session"); + final ClientSession session = sf.createSession(false, false); + log.info("created session"); + + session.createQueue(ADDRESS, ADDRESS, null, true); + + final CountDownLatch latch = new CountDownLatch(1); + + class MyListener implements FailureListener + { + public void connectionFailed(HornetQException me) + { + latch.countDown(); + } + } + + session.addFailureListener(new MyListener()); + + final int numMessages = 100; + + ClientProducer producer = session.createProducer(ADDRESS); + + String txID = "my-tx-id"; + + for (int i = 0; i < numMessages; i++) + { + ClientMessage message = session.createClientMessage(true); + + if (i == 0) + { + // Only need to add it on one message per tx + message.putStringProperty(MessageImpl.HDR_DUPLICATE_DETECTION_ID, new SimpleString(txID)); + } + + message.getBody().writeString("message" + i); + + message.putIntProperty("counter", i); + + producer.send(message); + } + + class Committer extends Thread + { + public void run() + { + Interceptor interceptor = new DelayInterceptor2(); + + try + { + sf.addInterceptor(interceptor); + + session.commit(); + + log.info("Initial commit succeeded"); + } + catch (HornetQException e) + { + if (e.getCode() == HornetQException.UNBLOCKED) + { + log.info("commit unblocked"); + + // Ok - now we retry the commit after removing the interceptor + + sf.removeInterceptor(interceptor); + + try + { + log.info("retrying commit"); + session.commit(); + } + catch (HornetQException e2) + { + if (e2.getCode() == HornetQException.TRANSACTION_ROLLED_BACK) + { + // Ok + + failed = false; + } + } + } + } + } + + volatile boolean failed = true; + } + + Committer committer = new Committer(); + + committer.start(); + + Thread.sleep(500); + + RemotingConnection conn = ((ClientSessionInternal)session).getConnection(); + + conn.fail(new HornetQException(HornetQException.NOT_CONNECTED)); + + // Wait to be informed of failure + + boolean ok = latch.await(1000, TimeUnit.MILLISECONDS); + + assertTrue(ok); + + committer.join(); + + assertFalse(committer.failed); + + session.close(); + + ClientSession session2 = sf.createSession(false, false); + + producer = session2.createProducer(ADDRESS); + + // We now try and resend the messages since we get a transaction rolled back exception + // but the commit actually succeeded, duplicate detection should kick in and prevent dups + + for (int i = 0; i < numMessages; i++) + { + ClientMessage message = session2.createClientMessage(true); + + if (i == 0) + { + // Only need to add it on one message per tx + message.putStringProperty(MessageImpl.HDR_DUPLICATE_DETECTION_ID, new SimpleString(txID)); + } + + message.getBody().writeString("message" + i); + + message.putIntProperty("counter", i); + + producer.send(message); + } + + session2.commit(); + + ClientConsumer consumer = session2.createConsumer(ADDRESS); + + session2.start(); + + for (int i = 0; i < numMessages; i++) + { + ClientMessage message = consumer.receive(1000); + + assertNotNull(message); + + assertEquals("message" + i, message.getBody().readString()); + + assertEquals(i, message.getProperty("counter")); + + message.acknowledge(); + } + + ClientMessage message = consumer.receive(500); + + assertNull(message); + + session2.close(); + + assertEquals(0, sf.numSessions()); + + assertEquals(0, sf.numConnections()); + } + + public void testCommitDidNotOccurUnblockedAndResend() throws Exception + { + ClientSessionFactoryInternal sf = this.getSessionFactory(); + + sf.setBlockOnNonPersistentSend(true); + sf.setBlockOnPersistentSend(true); + sf.setBlockOnAcknowledge(true); + + log.info("creating session"); + final ClientSession session = sf.createSession(false, false); + log.info("created session"); + + session.createQueue(ADDRESS, ADDRESS, null, true); + + final CountDownLatch latch = new CountDownLatch(1); + + class MyListener implements FailureListener + { + public void connectionFailed(HornetQException me) + { + latch.countDown(); + } + } + + session.addFailureListener(new MyListener()); + + final int numMessages = 100; + + ClientProducer producer = session.createProducer(ADDRESS); + + for (int i = 0; i < numMessages; i++) + { + ClientMessage message = session.createClientMessage(true); + + message.getBody().writeString("message" + i); + + message.putIntProperty("counter", i); + + producer.send(message); + } + + class Committer extends Thread + { + public void run() + { + Interceptor interceptor = new DelayInterceptor3(); + + try + { + server0Service.getRemotingService().addInterceptor(interceptor); + + session.commit(); + + log.info("Initial commit succeeded"); + } + catch (HornetQException e) + { + if (e.getCode() == HornetQException.UNBLOCKED) + { + log.info("commit unblocked"); + + // Ok - now we retry the commit after removing the interceptor + + server0Service.getRemotingService().removeInterceptor(interceptor); + + try + { + session.commit(); + } + catch (HornetQException e2) + { + if (e2.getCode() == HornetQException.TRANSACTION_ROLLED_BACK) + { + // Ok + + failed = false; + } + } + } + } + } + + volatile boolean failed = true; + } + + Committer committer = new Committer(); + + committer.start(); + + Thread.sleep(500); + + RemotingConnection conn = ((ClientSessionInternal)session).getConnection(); + + conn.fail(new HornetQException(HornetQException.NOT_CONNECTED)); + + // Wait to be informed of failure + + boolean ok = latch.await(1000, TimeUnit.MILLISECONDS); + + assertTrue(ok); + + committer.join(); + + assertFalse(committer.failed); + + session.close(); + + ClientSession session2 = sf.createSession(false, false); + + producer = session2.createProducer(ADDRESS); + + // We now try and resend the messages since we get a transaction rolled back exception + + for (int i = 0; i < numMessages; i++) + { + ClientMessage message = session2.createClientMessage(true); + + message.getBody().writeString("message" + i); + + message.putIntProperty("counter", i); + + producer.send(message); + } + + session2.commit(); + + ClientConsumer consumer = session2.createConsumer(ADDRESS); + + session2.start(); + + for (int i = 0; i < numMessages; i++) + { + ClientMessage message = consumer.receive(1000); + + assertNotNull(message); + + assertEquals("message" + i, message.getBody().readString()); + + assertEquals(i, message.getProperty("counter")); + + message.acknowledge(); + } + + ClientMessage message = consumer.receive(500); + + assertNull(message); + + session2.close(); + + assertEquals(0, sf.numSessions()); + + assertEquals(0, sf.numConnections()); + } + + // Package protected --------------------------------------------- + + // Protected ----------------------------------------------------- + + @Override + protected TransportConfiguration getAcceptorTransportConfiguration(boolean live) + { + if (live) + { + return new TransportConfiguration("org.hornetq.core.remoting.impl.invm.InVMAcceptorFactory"); + } + else + { + Map<String, Object> server1Params = new HashMap<String, Object>(); + + server1Params.put(TransportConstants.SERVER_ID_PROP_NAME, 1); + + return new TransportConfiguration("org.hornetq.core.remoting.impl.invm.InVMAcceptorFactory", server1Params); + } + } + + @Override + protected TransportConfiguration getConnectorTransportConfiguration(final boolean live) + { + if (live) + { + return new TransportConfiguration("org.hornetq.core.remoting.impl.invm.InVMConnectorFactory"); + } + else + { + Map<String, Object> server1Params = new HashMap<String, Object>(); + + server1Params.put(TransportConstants.SERVER_ID_PROP_NAME, 1); + + return new TransportConfiguration("org.hornetq.core.remoting.impl.invm.InVMConnectorFactory", server1Params); + } + } + + // Private ------------------------------------------------------- + + private ClientSession sendAndConsume(final ClientSessionFactory sf) throws Exception + { + ClientSession session = sf.createSession(false, true, true); + + session.createQueue(ADDRESS, ADDRESS, null, false); + + ClientProducer producer = session.createProducer(ADDRESS); + + final int numMessages = 1000; + + for (int i = 0; i < numMessages; i++) + { + ClientMessage message = session.createClientMessage(HornetQTextMessage.TYPE, + false, + 0, + System.currentTimeMillis(), + (byte)1); + message.putIntProperty(new SimpleString("count"), i); + message.getBody().writeString("aardvarks"); + producer.send(message); + } + + ClientConsumer consumer = session.createConsumer(ADDRESS); + + session.start(); + + for (int i = 0; i < numMessages; i++) + { + ClientMessage message2 = consumer.receive(); + + assertEquals("aardvarks", message2.getBody().readString()); + + assertEquals(i, message2.getProperty(new SimpleString("count"))); + + message2.acknowledge(); + } + + ClientMessage message3 = consumer.receive(250); + + assertNull(message3); + + return session; + } + // Inner classes ------------------------------------------------- } Added: branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/integration/cluster/failover/FailoverTestBase.java =================================================================== --- branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/integration/cluster/failover/FailoverTestBase.java (rev 0) +++ branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/integration/cluster/failover/FailoverTestBase.java 2009-09-22 15:08:41 UTC (rev 7979) @@ -0,0 +1,111 @@ +/* + * Copyright 2009 Red Hat, Inc. + * Red Hat licenses this file to you under the Apache License, version + * 2.0 (the "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * http://www.apache.org/licenses/LICENSE-2.0 + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or + * implied. See the License for the specific language governing + * permissions and limitations under the License. + */ + +package org.hornetq.tests.integration.cluster.failover; + +import org.hornetq.core.client.impl.ClientSessionFactoryImpl; +import org.hornetq.core.client.impl.ClientSessionFactoryInternal; +import org.hornetq.core.config.Configuration; +import org.hornetq.core.config.TransportConfiguration; +import org.hornetq.core.remoting.impl.invm.InVMConnector; +import org.hornetq.core.remoting.impl.invm.InVMRegistry; +import org.hornetq.core.server.HornetQServer; +import org.hornetq.tests.util.ServiceTestBase; +import org.hornetq.utils.SimpleString; + +/** + * A FailoverTestBase + * + * @author tim + * + * + */ +public abstract class FailoverTestBase extends ServiceTestBase +{ + // Constants ----------------------------------------------------- + + protected static final SimpleString ADDRESS = new SimpleString("FailoverTestAddress"); + + // Attributes ---------------------------------------------------- + + protected HornetQServer server0Service; + + protected HornetQServer server1Service; + + // Static -------------------------------------------------------- + + // Constructors -------------------------------------------------- + + // Public -------------------------------------------------------- + + // Package protected --------------------------------------------- + + // Protected ----------------------------------------------------- + + protected void setUp() throws Exception + { + super.setUp(); + + Configuration config1 = super.createDefaultConfig(); + config1.getAcceptorConfigurations().clear(); + config1.getAcceptorConfigurations() + .add(getAcceptorTransportConfiguration(false)); + config1.setSecurityEnabled(false); + config1.setSharedStore(true); + config1.setBackup(true); + server1Service = super.createServer(true, config1); + + Configuration config0 = super.createDefaultConfig(); + config0.getAcceptorConfigurations().clear(); + config0.getAcceptorConfigurations() + .add(getAcceptorTransportConfiguration(true)); + config0.setSecurityEnabled(false); + config0.setSharedStore(true); + server0Service = super.createServer(true, config0); + + server1Service.start(); + server0Service.start(); + } + + + protected void tearDown() throws Exception + { + server1Service.stop(); + + server0Service.stop(); + + assertEquals(0, InVMRegistry.instance.size()); + + server1Service = null; + + server0Service = null; + + InVMConnector.failOnCreateConnection = false; + + super.tearDown(); + } + + protected abstract TransportConfiguration getAcceptorTransportConfiguration(boolean live); + + protected abstract TransportConfiguration getConnectorTransportConfiguration(final boolean live); + + protected ClientSessionFactoryInternal getSessionFactory() + { + return new ClientSessionFactoryImpl(getConnectorTransportConfiguration(true), getConnectorTransportConfiguration(false)); + } + + // Private ------------------------------------------------------- + + // Inner classes ------------------------------------------------- + +} Deleted: branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/integration/cluster/failover/FailureListenerOnFailoverTest.java =================================================================== --- branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/integration/cluster/failover/FailureListenerOnFailoverTest.java 2009-09-22 13:57:55 UTC (rev 7978) +++ branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/integration/cluster/failover/FailureListenerOnFailoverTest.java 2009-09-22 15:08:41 UTC (rev 7979) @@ -1,559 +0,0 @@ -/* - * Copyright 2009 Red Hat, Inc. - * Red Hat licenses this file to you under the Apache License, version - * 2.0 (the "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * http://www.apache.org/licenses/LICENSE-2.0 - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or - * implied. See the License for the specific language governing - * permissions and limitations under the License. - */ - -package org.hornetq.tests.integration.cluster.failover; - -import java.util.ArrayList; -import java.util.HashMap; -import java.util.HashSet; -import java.util.List; -import java.util.Map; -import java.util.Set; - -import org.hornetq.core.client.ClientSession; -import org.hornetq.core.client.impl.ClientSessionFactoryImpl; -import org.hornetq.core.client.impl.ClientSessionFactoryInternal; -import org.hornetq.core.client.impl.ClientSessionInternal; -import org.hornetq.core.config.Configuration; -import org.hornetq.core.config.TransportConfiguration; -import org.hornetq.core.config.impl.ConfigurationImpl; -import org.hornetq.core.exception.HornetQException; -import org.hornetq.core.logging.Logger; -import org.hornetq.core.remoting.FailureListener; -import org.hornetq.core.remoting.RemotingConnection; -import org.hornetq.core.remoting.impl.invm.InVMConnector; -import org.hornetq.core.remoting.impl.invm.InVMRegistry; -import org.hornetq.core.remoting.impl.invm.TransportConstants; -import org.hornetq.core.server.HornetQ; -import org.hornetq.core.server.HornetQServer; -import org.hornetq.tests.util.UnitTestCase; - -/** - * - * A FailureListenerOnFailoverTest - * - * Make sure FailuerListener is called at the right places during the failover process - * - * @author <a href="mailto:tim.fox@jboss.com">Tim Fox</a> - * - * Created 4 Nov 2008 16:54:50 - * - * - */ -public class FailureListenerOnFailoverTest extends UnitTestCase -{ - private static final Logger log = Logger.getLogger(FailureListenerOnFailoverTest.class); - - // Constants ----------------------------------------------------- - - // Attributes ---------------------------------------------------- - - private HornetQServer liveService; - - private HornetQServer backupService; - - private Map<String, Object> backupParams = new HashMap<String, Object>(); - - // Static -------------------------------------------------------- - - // Constructors -------------------------------------------------- - - // Public -------------------------------------------------------- - - class MyListener implements FailureListener - { - private int i; - - MyListener(int i) - { - this.i = i; - } - - int failCount; - - public synchronized void connectionFailed(final HornetQException me) - { - failCount++; - } - - synchronized int getFailCount() - { - return failCount; - } - } - - /* - * Listeners shouldn't be called if failed over successfully - */ - public void testFailureListenersNotCalledOnFailover() throws Exception - { - ClientSessionFactoryInternal sf = new ClientSessionFactoryImpl(new TransportConfiguration("org.hornetq.core.remoting.impl.invm.InVMConnectorFactory"), - new TransportConfiguration("org.hornetq.core.remoting.impl.invm.InVMConnectorFactory", - backupParams)); - - final int numSessions = (int)(ClientSessionFactoryImpl.DEFAULT_MAX_CONNECTIONS * 1.5); - - List<MyListener> listeners = new ArrayList<MyListener>(); - - RemotingConnection conn = null; - - Set<ClientSession> sessions = new HashSet<ClientSession>(); - - for (int i = 0; i < numSessions; i++) - { - ClientSession session = sf.createSession(false, true, true); - - if (conn == null) - { - conn = ((ClientSessionInternal)session).getConnection(); - } - - MyListener listener = new MyListener(i); - - session.addFailureListener(listener); - - listeners.add(listener); - - sessions.add(session); - } - - conn.fail(new HornetQException(HornetQException.NOT_CONNECTED)); - - for (MyListener listener : listeners) - { - assertEquals(0, listener.getFailCount()); - } - - // Do some stuff to make sure sessions failed over/reconnected ok - int i = 0; - for (ClientSession session : sessions) - { - session.createQueue("testaddress" + i, "testaddress" + i, false); - session.deleteQueue("testaddress" + i); - i++; - session.close(); - } - - sf.close(); - } - - /* - * Listeners shouldn't be called if reconnected successfully - */ - public void testFailureListenersNotCalledOnReconnection() throws Exception - { - final long retryInterval = 10; - - final double retryMultiplier = 1d; - - final int reconnectAttempts = 10; - - ClientSessionFactoryInternal sf = new ClientSessionFactoryImpl(new TransportConfiguration("org.hornetq.core.remoting.impl.invm.InVMConnectorFactory")); - - sf.setRetryInterval(retryInterval); - sf.setRetryIntervalMultiplier(retryMultiplier); - sf.setReconnectAttempts(reconnectAttempts); - - final int numSessions = (int)(ClientSessionFactoryImpl.DEFAULT_MAX_CONNECTIONS * 1.5); - - List<MyListener> listeners = new ArrayList<MyListener>(); - - RemotingConnection conn = null; - - Set<ClientSession> sessions = new HashSet<ClientSession>(); - - for (int i = 0; i < numSessions; i++) - { - ClientSession session = sf.createSession(false, true, true); - - if (conn == null) - { - conn = ((ClientSessionInternal)session).getConnection(); - } - - MyListener listener = new MyListener(i); - - session.addFailureListener(listener); - - listeners.add(listener); - - sessions.add(session); - } - - conn.fail(new HornetQException(HornetQException.NOT_CONNECTED)); - - for (MyListener listener : listeners) - { - assertEquals(0, listener.getFailCount()); - } - - try - { - // Do some stuff to make sure sessions failed over/reconnected ok - int i = 0; - for (ClientSession session : sessions) - { - session.createQueue("testaddress" + i, "testaddress" + i, false); - session.deleteQueue("testaddress" + i); - i++; - session.close(); - } - } - finally - { - sf.close(); - } - } - - /* - * Listeners should be called if no backup server - */ - public void testFailureListenerCalledNoBackup() throws Exception - { - ClientSessionFactoryInternal sf = new ClientSessionFactoryImpl(new TransportConfiguration("org.hornetq.core.remoting.impl.invm.InVMConnectorFactory")); - - final int numSessions = (int)(ClientSessionFactoryImpl.DEFAULT_MAX_CONNECTIONS * 1.5); - - List<MyListener> listeners = new ArrayList<MyListener>(); - - RemotingConnection conn = null; - - Set<ClientSession> sessions = new HashSet<ClientSession>(); - - for (int i = 0; i < numSessions; i++) - { - ClientSession session = sf.createSession(false, true, true); - - sessions.add(session); - - if (conn == null) - { - conn = ((ClientSessionInternal)session).getConnection(); - } - - MyListener listener = new MyListener(i); - - session.addFailureListener(listener); - - listeners.add(listener); - } - - conn.fail(new HornetQException(HornetQException.NOT_CONNECTED)); - - for (MyListener listener : listeners) - { - assertEquals(1, listener.getFailCount()); - } - - for (ClientSession session : sessions) - { - session.close(); - } - - sf.close(); - } - - /* - * Listener should be called if failed to reconnect, no backup present - */ - public void testFailureListenerCalledOnFailureToReconnect() throws Exception - { - final long retryInterval = 10; - - final double retryMultiplier = 1d; - - final int reconnectAttempts = 10; - - ClientSessionFactoryInternal sf = new ClientSessionFactoryImpl(new TransportConfiguration("org.hornetq.core.remoting.impl.invm.InVMConnectorFactory")); - - sf.setRetryInterval(retryInterval); - sf.setRetryIntervalMultiplier(retryMultiplier); - sf.setReconnectAttempts(reconnectAttempts); - - final int numSessions = (int)(ClientSessionFactoryImpl.DEFAULT_MAX_CONNECTIONS * 1.5); - - List<MyListener> listeners = new ArrayList<MyListener>(); - - RemotingConnection conn = null; - - Set<ClientSession> sessions = new HashSet<ClientSession>(); - - for (int i = 0; i < numSessions; i++) - { - ClientSession session = sf.createSession(false, true, true); - - sessions.add(session); - - if (conn == null) - { - conn = ((ClientSessionInternal)session).getConnection(); - } - - MyListener listener = new MyListener(i); - - session.addFailureListener(listener); - - listeners.add(listener); - } - - InVMConnector.failOnCreateConnection = true; - - conn.fail(new HornetQException(HornetQException.NOT_CONNECTED)); - - int i = 0; - for (MyListener listener : listeners) - { - assertEquals(1, listener.getFailCount()); - } - - for (ClientSession session : sessions) - { - session.close(); - } - - sf.close(); - } - - /* - * Listener should be called if failed to reconnect after failover, backup present - */ - public void testFailureListenerCalledOnFailureToReconnectBackupPresent() throws Exception - { - final long retryInterval = 10; - - final double retryMultiplier = 1d; - - final int reconnectAttempts = 10; - - ClientSessionFactoryInternal sf = new ClientSessionFactoryImpl(new TransportConfiguration("org.hornetq.core.remoting.impl.invm.InVMConnectorFactory"), - new TransportConfiguration("org.hornetq.core.remoting.impl.invm.InVMConnectorFactory", - backupParams)); - - sf.setFailoverOnServerShutdown(true); - sf.setRetryInterval(retryInterval); - sf.setRetryIntervalMultiplier(retryMultiplier); - sf.setReconnectAttempts(reconnectAttempts); - - final int numSessions = (int)(ClientSessionFactoryImpl.DEFAULT_MAX_CONNECTIONS * 1.5); - - List<MyListener> listeners = new ArrayList<MyListener>(); - - RemotingConnection conn = null; - - Set<ClientSession> sessions = new HashSet<ClientSession>(); - - for (int i = 0; i < numSessions; i++) - { - ClientSession session = sf.createSession(false, true, true); - - if (conn == null) - { - conn = ((ClientSessionInternal)session).getConnection(); - } - - MyListener listener = new MyListener(i); - - session.addFailureListener(listener); - - listeners.add(listener); - - sessions.add(session); - } - - // Fail once to failover ok - - conn.fail(new HornetQException(HornetQException.NOT_CONNECTED)); - - for (MyListener listener : listeners) - { - assertEquals(0, listener.getFailCount()); - } - - // Do some stuff to make sure sessions failed over/reconnected ok - int i = 0; - for (ClientSession session : sessions) - { - session.createQueue("testaddress" + i, "testaddress" + i, false); - session.deleteQueue("testaddress" + i); - i++; - } - - // Now fail again and reconnect ok - - ClientSession csession = sf.createSession(false, true, true); - - conn = ((ClientSessionInternal)csession).getConnection(); - - InVMConnector.failOnCreateConnection = true; - InVMConnector.numberOfFailures = reconnectAttempts - 1; - - conn.fail(new HornetQException(HornetQException.NOT_CONNECTED)); - - i = 0; - for (ClientSession session : sessions) - { - session.createQueue("testaddress" + i, "testaddress" + i, false); - session.deleteQueue("testaddress" + i); - i++; - } - - csession.close(); - - // Now fail again and fail to reconnect - - csession = sf.createSession(false, true, true); - - conn = ((ClientSessionInternal)csession).getConnection(); - - InVMConnector.failOnCreateConnection = true; - InVMConnector.numberOfFailures = -1; - - conn.fail(new HornetQException(HornetQException.NOT_CONNECTED)); - - i = 0; - for (MyListener listener : listeners) - { - assertEquals(1, listener.getFailCount()); - } - - csession.close(); - - for (ClientSession session : sessions) - { - session.close(); - } - - sf.close(); - } - - /* - * Listener should be called if failed to failover - */ - public void testFailureListenerCalledOnFailureToFailover() throws Exception - { - final long retryInterval = 10; - - final double retryMultiplier = 1d; - - final int reconnectAttempts = 1; - - ClientSessionFactoryInternal sf = new ClientSessionFactoryImpl(new TransportConfiguration("org.hornetq.core.remoting.impl.invm.InVMConnectorFactory"), - new TransportConfiguration("org.hornetq.core.remoting.impl.invm.InVMConnectorFactory", - backupParams)); - - sf.setFailoverOnServerShutdown(true); - sf.setRetryInterval(retryInterval); - sf.setRetryIntervalMultiplier(retryMultiplier); - sf.setReconnectAttempts(reconnectAttempts); - - final int numSessions = (int)(ClientSessionFactoryImpl.DEFAULT_MAX_CONNECTIONS * 1.5); - - List<MyListener> listeners = new ArrayList<MyListener>(); - - RemotingConnection conn = null; - - Set<ClientSession> sessions = new HashSet<ClientSession>(); - - for (int i = 0; i < numSessions; i++) - { - ClientSession session = sf.createSession(false, true, true); - - sessions.add(session); - - if (conn == null) - { - conn = ((ClientSessionInternal)session).getConnection(); - } - - MyListener listener = new MyListener(i); - - session.addFailureListener(listener); - - listeners.add(listener); - } - - InVMConnector.failOnCreateConnection = true; - - conn.fail(new HornetQException(HornetQException.NOT_CONNECTED)); - - for (MyListener listener : listeners) - { - assertEquals(1, listener.getFailCount()); - } - - for (ClientSession session: sessions) - { - session.close(); - } - - sf.close(); - } - - // Package protected --------------------------------------------- - - // Protected ----------------------------------------------------- - - @Override - protected void setUp() throws Exception - { - super.setUp(); - - Configuration backupConf = new ConfigurationImpl(); - backupConf.setSecurityEnabled(false); - backupParams.put(TransportConstants.SERVER_ID_PROP_NAME, 1); - backupConf.getAcceptorConfigurations() - .add(new TransportConfiguration("org.hornetq.core.remoting.impl.invm.InVMAcceptorFactory", - backupParams)); - backupConf.setBackup(true); - backupService = HornetQ.newHornetQServer(backupConf, false); - backupService.start(); - - Configuration liveConf = new ConfigurationImpl(); - liveConf.setSecurityEnabled(false); - liveConf.getAcceptorConfigurations() - .add(new TransportConfiguration("org.hornetq.core.remoting.impl.invm.InVMAcceptorFactory")); - Map<String, TransportConfiguration> connectors = new HashMap<String, TransportConfiguration>(); - TransportConfiguration backupTC = new TransportConfiguration("org.hornetq.core.remoting.impl.invm.InVMConnectorFactory", - backupParams, - "backup-connector"); - connectors.put(backupTC.getName(), backupTC); - liveConf.setConnectorConfigurations(connectors); - liveConf.setBackupConnectorName(backupTC.getName()); - liveService = HornetQ.newHornetQServer(liveConf, false); - liveService.start(); - } - - @Override - protected void tearDown() throws Exception - { - InVMConnector.resetFailures(); - - backupService.stop(); - - liveService.stop(); - - assertEquals(0, InVMRegistry.instance.size()); - - backupService = null; - - liveService = null; - - backupParams = null; - - super.tearDown(); - } - - // Private ------------------------------------------------------- - - // Inner classes ------------------------------------------------- -} Deleted: branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/integration/cluster/failover/MultiThreadFailoverSupport.java =================================================================== --- branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/integration/cluster/failover/MultiThreadFailoverSupport.java 2009-09-22 13:57:55 UTC (rev 7978) +++ branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/integration/cluster/failover/MultiThreadFailoverSupport.java 2009-09-22 15:08:41 UTC (rev 7979) @@ -1,278 +0,0 @@ -/* - * Copyright 2009 Red Hat, Inc. - * Red Hat licenses this file to you under the Apache License, version - * 2.0 (the "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * http://www.apache.org/licenses/LICENSE-2.0 - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or - * implied. See the License for the specific language governing - * permissions and limitations under the License. - */ - - -package org.hornetq.tests.integration.cluster.failover; - -import java.util.ArrayList; -import java.util.List; -import java.util.Timer; -import java.util.TimerTask; - -import org.hornetq.core.client.ClientSession; -import org.hornetq.core.client.ClientSessionFactory; -import org.hornetq.core.client.impl.ClientSessionFactoryInternal; -import org.hornetq.core.client.impl.ClientSessionInternal; -import org.hornetq.core.exception.HornetQException; -import org.hornetq.core.logging.Logger; -import org.hornetq.core.remoting.impl.RemotingConnectionImpl; -import org.hornetq.core.remoting.impl.invm.InVMConnector; -import org.hornetq.tests.util.ServiceTestBase; - -/** - * A MultiThreadFailoverSupport - * - * @author <a href="mailto:time.fox@jboss.org">Tim Fox</a> - * @author <a href="mailto:clebert.suconic@jboss.org">Clebert Suconic</a> - * - * Created Mar 17, 2009 11:15:02 AM - * - * - */ -public abstract class MultiThreadFailoverSupport extends ServiceTestBase -{ - - // Constants ----------------------------------------------------- - - private final Logger log = Logger.getLogger(this.getClass()); - - // Attributes ---------------------------------------------------- - - private Timer timer; - - // Static -------------------------------------------------------- - - // Constructors -------------------------------------------------- - - // Public -------------------------------------------------------- - - // Package protected --------------------------------------------- - - // Protected ----------------------------------------------------- - - protected abstract void start() throws Exception; - - protected abstract void stop() throws Exception; - - protected abstract ClientSessionFactoryInternal createSessionFactory(); - - protected void setUp() throws Exception - { - super.setUp(); - timer = new Timer(); - } - - protected void tearDown() throws Exception - { - timer.cancel(); - timer = null; - super.tearDown(); - } - - protected boolean shouldFail() - { - return true; - } - - protected void runMultipleThreadsFailoverTest(final RunnableT runnable, - final int numThreads, - final int numIts, - final boolean failOnCreateConnection, - final long failDelay) throws Exception - { - for (int its = 0; its < numIts; its++) - { - log.info("Beginning iteration " + its); - - start(); - - final ClientSessionFactoryInternal sf = createSessionFactory(); - - final ClientSession session = sf.createSession(false, true, true); - - Failer failer = startFailer(failDelay, session, failOnCreateConnection); - - class Runner extends Thread - { - private volatile Throwable throwable; - - private final RunnableT test; - - private final int threadNum; - - Runner(final RunnableT test, final int threadNum) - { - this.test = test; - - this.threadNum = threadNum; - } - - @Override - public void run() - { - try - { - test.run(sf, threadNum); - } - catch (Throwable t) - { - throwable = t; - - log.error("Failed to run test", t); - - // Case a failure happened here, it should print the Thread dump - // Sending it to System.out, as it would show on the Tests report - System.out.println(threadDump(" - fired by MultiThreadRandomReattachTestBase::runTestMultipleThreads (" + t.getLocalizedMessage() + ")")); - } - } - } - - do - { - List<Runner> threads = new ArrayList<Runner>(); - - for (int i = 0; i < numThreads; i++) - { - Runner runner = new Runner(runnable, i); - - threads.add(runner); - - runner.start(); - } - - for (Runner thread : threads) - { - thread.join(); - - if (thread.throwable != null) - { - throw new Exception("Exception on thread " + thread, thread.throwable); - } - } - - runnable.checkFail(); - - } - while (!failer.isExecuted()); - - InVMConnector.resetFailures(); - - session.close(); - - assertEquals(0, sf.numSessions()); - - assertEquals(0, sf.numConnections()); - - sf.close(); - - stop(); - } - } - - - // Private ------------------------------------------------------- - - private Failer startFailer(final long time, final ClientSession session, final boolean failOnCreateConnection) - { - Failer failer = new Failer(session, failOnCreateConnection); - - // This is useful for debugging.. just change shouldFail to return false, and Failer will not be executed - if (shouldFail()) - { - timer.schedule(failer, (long)(time * Math.random()), 100); - } - - return failer; - } - - - // Inner classes ------------------------------------------------- - - - protected abstract class RunnableT extends Thread - { - private volatile String failReason; - - private volatile Throwable throwable; - - public void setFailed(final String reason, final Throwable throwable) - { - failReason = reason; - this.throwable = throwable; - } - - public void checkFail() - { - if (throwable != null) - { - log.error("Test failed: " + failReason, throwable); - } - if (failReason != null) - { - fail(failReason); - } - } - - public abstract void run(final ClientSessionFactory sf, final int threadNum) throws Exception; - } - - - - private class Failer extends TimerTask - { - private final ClientSession session; - - private boolean executed; - - private final boolean failOnCreateConnection; - - public Failer(final ClientSession session, final boolean failOnCreateConnection) - { - this.session = session; - - this.failOnCreateConnection = failOnCreateConnection; - } - - @Override - public synchronized void run() - { - log.info("** Failing connection"); - - RemotingConnectionImpl conn = (RemotingConnectionImpl)((ClientSessionInternal)session).getConnection(); - - if (failOnCreateConnection) - { - InVMConnector.numberOfFailures = 1; - InVMConnector.failOnCreateConnection = true; - } - else - { - conn.fail(new HornetQException(HornetQException.NOT_CONNECTED, "blah")); - } - - log.info("** Fail complete"); - - cancel(); - - executed = true; - } - - public synchronized boolean isExecuted() - { - return executed; - } - } - - - -} Deleted: branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/integration/cluster/failover/MultiThreadRandomReattachTest.java =================================================================== --- branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/integration/cluster/failover/MultiThreadRandomReattachTest.java 2009-09-22 13:57:55 UTC (rev 7978) +++ branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/integration/cluster/failover/MultiThreadRandomReattachTest.java 2009-09-22 15:08:41 UTC (rev 7979) @@ -1,69 +0,0 @@ -/* - * Copyright 2009 Red Hat, Inc. - * Red Hat licenses this file to you under the Apache License, version - * 2.0 (the "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * http://www.apache.org/licenses/LICENSE-2.0 - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or - * implied. See the License for the specific language governing - * permissions and limitations under the License. - */ - -package org.hornetq.tests.integration.cluster.failover; - -import java.util.HashMap; -import java.util.Map; - -import org.hornetq.core.client.ClientMessage; -import org.hornetq.core.config.Configuration; -import org.hornetq.core.config.TransportConfiguration; -import org.hornetq.core.config.impl.ConfigurationImpl; -import org.hornetq.core.server.HornetQ; - - -/** - * - * A MultiThreadRandomReattachTest - * - * @author <a href="mailto:tim.fox@jboss.com">Tim Fox</a> - * - * - */ -public class MultiThreadRandomReattachTest extends MultiThreadRandomReattachTestBase -{ - @Override - protected void start() throws Exception - { - Configuration liveConf = new ConfigurationImpl(); - liveConf.setSecurityEnabled(false); - liveConf.getAcceptorConfigurations() - .add(new TransportConfiguration("org.hornetq.core.remoting.impl.invm.InVMAcceptorFactory")); - Map<String, TransportConfiguration> connectors = new HashMap<String, TransportConfiguration>(); - TransportConfiguration backupTC = new TransportConfiguration("org.hornetq.core.remoting.impl.invm.InVMConnectorFactory"); - connectors.put(backupTC.getName(), backupTC); - liveConf.setConnectorConfigurations(connectors); - liveConf.setBackupConnectorName(backupTC.getName()); - liveServer = HornetQ.newHornetQServer(liveConf, false); - liveServer.start(); - } - - /* (non-Javadoc) - * @see org.hornetq.tests.integration.cluster.failover.MultiThreadRandomReattachTestBase#setBody(org.hornetq.core.client.ClientMessage) - */ - @Override - protected void setBody(final ClientMessage message) throws Exception - { - } - - /* (non-Javadoc) - * @see org.hornetq.tests.integration.cluster.failover.MultiThreadRandomReattachTestBase#checkSize(org.hornetq.core.client.ClientMessage) - */ - @Override - protected boolean checkSize(final ClientMessage message) - { - return 0 == message.getBody().writerIndex(); - } - -} Deleted: branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/integration/cluster/failover/MultiThreadRandomReattachTestBase.java =================================================================== --- branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/integration/cluster/failover/MultiThreadRandomReattachTestBase.java 2009-09-22 13:57:55 UTC (rev 7978) +++ branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/integration/cluster/failover/MultiThreadRandomReattachTestBase.java 2009-09-22 15:08:41 UTC (rev 7979) @@ -1,1500 +0,0 @@ -/* - * Copyright 2009 Red Hat, Inc. - * Red Hat licenses this file to you under the Apache License, version - * 2.0 (the "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * http://www.apache.org/licenses/LICENSE-2.0 - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or - * implied. See the License for the specific language governing - * permissions and limitations under the License. - */ - -package org.hornetq.tests.integration.cluster.failover; - -import java.util.HashMap; -import java.util.HashSet; -import java.util.Map; -import java.util.Set; -import java.util.concurrent.CountDownLatch; -import java.util.concurrent.TimeUnit; - -import org.hornetq.core.client.ClientConsumer; -import org.hornetq.core.client.ClientMessage; -import org.hornetq.core.client.ClientProducer; -import org.hornetq.core.client.ClientSession; -import org.hornetq.core.client.ClientSessionFactory; -import org.hornetq.core.client.MessageHandler; -import org.hornetq.core.client.impl.ClientSessionFactoryImpl; -import org.hornetq.core.client.impl.ClientSessionFactoryInternal; -import org.hornetq.core.config.TransportConfiguration; -import org.hornetq.core.exception.HornetQException; -import org.hornetq.core.logging.Logger; -import org.hornetq.core.remoting.impl.invm.InVMRegistry; -import org.hornetq.core.server.HornetQServer; -import org.hornetq.jms.client.HornetQBytesMessage; -import org.hornetq.jms.client.HornetQTextMessage; -import org.hornetq.utils.SimpleString; - -/** - * A MultiThreadRandomReattachTestBase - * - * @author <a href="mailto:tim.fox@jboss.com">Tim Fox</a> - * @author <a href="mailto:clebert.suconic@jboss.org">Clebert Suconic</a> - * - * - */ -public abstract class MultiThreadRandomReattachTestBase extends MultiThreadFailoverSupport -{ - private final Logger log = Logger.getLogger(getClass()); - - // Constants ----------------------------------------------------- - - private static final int RECEIVE_TIMEOUT = 30000; - - private final int LATCH_WAIT = getLatchWait(); - - private int NUM_THREADS = getNumThreads(); - - // Attributes ---------------------------------------------------- - protected static final SimpleString ADDRESS = new SimpleString("FailoverTestAddress"); - - protected HornetQServer liveServer; - - // Static -------------------------------------------------------- - - // Constructors -------------------------------------------------- - - // Public -------------------------------------------------------- - - public void testA() throws Exception - { - runTestMultipleThreads(new RunnableT() - { - @Override - public void run(final ClientSessionFactory sf, final int threadNum) throws Exception - { - doTestA(sf, threadNum); - } - }, NUM_THREADS, false); - - } - - public void testB() throws Exception - { - runTestMultipleThreads(new RunnableT() - { - @Override - public void run(final ClientSessionFactory sf, final int threadNum) throws Exception - { - doTestB(sf, threadNum); - } - }, NUM_THREADS, false); - } - - public void testC() throws Exception - { - runTestMultipleThreads(new RunnableT() - { - @Override - public void run(final ClientSessionFactory sf, final int threadNum) throws Exception - { - doTestC(sf, threadNum); - } - }, NUM_THREADS, false); - } - - public void testD() throws Exception - { - runTestMultipleThreads(new RunnableT() - { - @Override - public void run(final ClientSessionFactory sf, final int threadNum) throws Exception - { - doTestD(sf, threadNum); - } - }, NUM_THREADS, false); - } - - public void testE() throws Exception - { - runTestMultipleThreads(new RunnableT() - { - @Override - public void run(final ClientSessionFactory sf, final int threadNum) throws Exception - { - doTestE(sf, threadNum); - } - }, NUM_THREADS, false); - } - - public void testF() throws Exception - { - runTestMultipleThreads(new RunnableT() - { - @Override - public void run(final ClientSessionFactory sf, final int threadNum) throws Exception - { - doTestF(sf, threadNum); - } - }, NUM_THREADS, false); - } - - public void testG() throws Exception - { - runTestMultipleThreads(new RunnableT() - { - @Override - public void run(final ClientSessionFactory sf, final int threadNum) throws Exception - { - doTestG(sf, threadNum); - } - }, NUM_THREADS, false); - } - - public void testH() throws Exception - { - runTestMultipleThreads(new RunnableT() - { - @Override - public void run(final ClientSessionFactory sf, final int threadNum) throws Exception - { - doTestH(sf, threadNum); - } - }, NUM_THREADS, false); - } - - public void testI() throws Exception - { - runTestMultipleThreads(new RunnableT() - { - @Override - public void run(final ClientSessionFactory sf, final int threadNum) throws Exception - { - doTestI(sf, threadNum); - } - }, NUM_THREADS, false); - } - - public void testJ() throws Exception - { - runTestMultipleThreads(new RunnableT() - { - @Override - public void run(final ClientSessionFactory sf, final int threadNum) throws Exception - { - doTestJ(sf, threadNum); - } - }, NUM_THREADS, false); - } - - public void testK() throws Exception - { - runTestMultipleThreads(new RunnableT() - { - @Override - public void run(final ClientSessionFactory sf, final int threadNum) throws Exception - { - doTestK(sf, threadNum); - } - }, NUM_THREADS, false); - } - - public void testL() throws Exception - { - runTestMultipleThreads(new RunnableT() - { - @Override - public void run(final ClientSessionFactory sf, final int threadNum) throws Exception - { - doTestL(sf); - } - }, NUM_THREADS, true, 10); - } - - // public void testM() throws Exception - // { - // runTestMultipleThreads(new RunnableT() - // { - // public void run(final ClientSessionFactory sf, final int threadNum) throws Exception - // { - // doTestM(sf, threadNum); - // } - // }, NUM_THREADS); - // } - - public void testN() throws Exception - { - runTestMultipleThreads(new RunnableT() - { - @Override - public void run(final ClientSessionFactory sf, final int threadNum) throws Exception - { - doTestN(sf, threadNum); - } - }, NUM_THREADS, false); - } - - // Package protected --------------------------------------------- - - // Protected ----------------------------------------------------- - - protected abstract void start() throws Exception; - - protected abstract void setBody(ClientMessage message) throws Exception; - - protected abstract boolean checkSize(ClientMessage message); - - protected int getNumThreads() - { - return 10; - } - - protected ClientSession createAutoCommitSession(ClientSessionFactory sf) throws Exception - { - return sf.createSession(false, true, true); - } - - protected ClientSession createTransactionalSession(ClientSessionFactory sf) throws Exception - { - return sf.createSession(false, false, false); - } - - protected void doTestA(final ClientSessionFactory sf, final int threadNum, final ClientSession session2) throws Exception - { - SimpleString subName = new SimpleString("sub" + threadNum); - - ClientSession session = sf.createSession(false, true, true); - - session.createQueue(ADDRESS, subName, null, false); - - ClientProducer producer = session.createProducer(ADDRESS); - - ClientConsumer consumer = session.createConsumer(subName); - - final int numMessages = 100; - - sendMessages(session, producer, numMessages, threadNum); - - session.start(); - - MyHandler handler = new MyHandler(threadNum, numMessages); - - consumer.setMessageHandler(handler); - - boolean ok = handler.latch.await(LATCH_WAIT, TimeUnit.MILLISECONDS); - - if (!ok) - { - throw new Exception("Timed out waiting for messages on handler " + System.identityHashCode(handler) + - " threadnum " + - threadNum); - } - - if (handler.failure != null) - { - throw new Exception("Handler failed: " + handler.failure); - } - - producer.close(); - - consumer.close(); - - session.deleteQueue(subName); - - session.close(); - } - - protected void doTestA(final ClientSessionFactory sf, final int threadNum) throws Exception - { - long start = System.currentTimeMillis(); - - ClientSession s = sf.createSession(false, false, false); - - final int numMessages = 100; - - final int numSessions = 10; - - Set<ClientConsumer> consumers = new HashSet<ClientConsumer>(); - Set<ClientSession> sessions = new HashSet<ClientSession>(); - - for (int i = 0; i < numSessions; i++) - { - SimpleString subName = new SimpleString(threadNum + "sub" + i); - - ClientSession sessConsume = createAutoCommitSession(sf); - - sessConsume.start(); - - sessConsume.createQueue(ADDRESS, subName, null, false); - - ClientConsumer consumer = sessConsume.createConsumer(subName); - - consumers.add(consumer); - - sessions.add(sessConsume); - - } - - ClientSession sessSend = sf.createSession(false, true, true); - - ClientProducer producer = sessSend.createProducer(ADDRESS); - - sendMessages(sessSend, producer, numMessages, threadNum); - - Set<MyHandler> handlers = new HashSet<MyHandler>(); - - for (ClientConsumer consumer : consumers) - { - MyHandler handler = new MyHandler(threadNum, numMessages); - - consumer.setMessageHandler(handler); - - handlers.add(handler); - } - - for (MyHandler handler : handlers) - { - boolean ok = handler.latch.await(LATCH_WAIT, TimeUnit.MILLISECONDS); - - if (!ok) - { - throw new Exception("Timed out waiting for messages on handler " + System.identityHashCode(handler) + - " threadnum " + - threadNum); - } - - if (handler.failure != null) - { - throw new Exception("Handler failed: " + handler.failure); - } - } - - sessSend.close(); - - for (ClientSession session : sessions) - { - session.close(); - } - - for (int i = 0; i < numSessions; i++) - { - SimpleString subName = new SimpleString(threadNum + "sub" + i); - - s.deleteQueue(subName); - } - - s.close(); - - long end = System.currentTimeMillis(); - - log.info("duration " + (end - start)); - } - - protected void doTestB(final ClientSessionFactory sf, final int threadNum) throws Exception - { - long start = System.currentTimeMillis(); - - ClientSession s = sf.createSession(false, false, false); - - final int numMessages = 100; - - final int numSessions = 10; - - Set<ClientConsumer> consumers = new HashSet<ClientConsumer>(); - Set<ClientSession> sessions = new HashSet<ClientSession>(); - - for (int i = 0; i < numSessions; i++) - { - SimpleString subName = new SimpleString(threadNum + "sub" + i); - - ClientSession sessConsume = createAutoCommitSession(sf); - - sessConsume.createQueue(ADDRESS, subName, null, false); - - ClientConsumer consumer = sessConsume.createConsumer(subName); - - consumers.add(consumer); - - sessions.add(sessConsume); - } - - ClientSession sessSend = sf.createSession(false, true, true); - - ClientProducer producer = sessSend.createProducer(ADDRESS); - - sendMessages(sessSend, producer, numMessages, threadNum); - - for (ClientSession session : sessions) - { - session.start(); - } - - Set<MyHandler> handlers = new HashSet<MyHandler>(); - - for (ClientConsumer consumer : consumers) - { - MyHandler handler = new MyHandler(threadNum, numMessages); - - consumer.setMessageHandler(handler); - - handlers.add(handler); - } - - for (MyHandler handler : handlers) - { - boolean ok = handler.latch.await(LATCH_WAIT, TimeUnit.MILLISECONDS); - - if (!ok) - { - throw new Exception("Timed out waiting for messages on handler " + System.identityHashCode(handler) + - " threadnum " + - threadNum); - } - - if (handler.failure != null) - { - throw new Exception("Handler failed: " + handler.failure); - } - } - - sessSend.close(); - - for (ClientSession session : sessions) - { - session.close(); - } - - for (int i = 0; i < numSessions; i++) - { - SimpleString subName = new SimpleString(threadNum + "sub" + i); - - s.deleteQueue(subName); - } - - s.close(); - - long end = System.currentTimeMillis(); - - log.info("duration " + (end - start)); - - } - - protected void doTestC(final ClientSessionFactory sf, final int threadNum) throws Exception - { - long start = System.currentTimeMillis(); - - ClientSession s = sf.createSession(false, false, false); - - final int numMessages = 100; - - final int numSessions = 10; - - Set<ClientConsumer> consumers = new HashSet<ClientConsumer>(); - Set<ClientSession> sessions = new HashSet<ClientSession>(); - - for (int i = 0; i < numSessions; i++) - { - SimpleString subName = new SimpleString(threadNum + "sub" + i); - - ClientSession sessConsume = createTransactionalSession(sf); - - sessConsume.start(); - - sessConsume.createQueue(ADDRESS, subName, null, false); - - ClientConsumer consumer = sessConsume.createConsumer(subName); - - consumers.add(consumer); - - sessions.add(sessConsume); - } - - ClientSession sessSend = sf.createSession(false, false, false); - - ClientProducer producer = sessSend.createProducer(ADDRESS); - - sendMessages(sessSend, producer, numMessages, threadNum); - - sessSend.rollback(); - - sendMessages(sessSend, producer, numMessages, threadNum); - - sessSend.commit(); - - Set<MyHandler> handlers = new HashSet<MyHandler>(); - - for (ClientConsumer consumer : consumers) - { - MyHandler handler = new MyHandler(threadNum, numMessages); - - consumer.setMessageHandler(handler); - - handlers.add(handler); - } - - for (MyHandler handler : handlers) - { - boolean ok = handler.latch.await(LATCH_WAIT, TimeUnit.MILLISECONDS); - - if (!ok) - { - throw new Exception("Timed out waiting for messages on handler " + System.identityHashCode(handler) + - " threadnum " + - threadNum); - } - - if (handler.failure != null) - { - throw new Exception("Handler failed: " + handler.failure); - } - - handler.reset(); - } - - for (ClientSession session : sessions) - { - session.rollback(); - } - - for (MyHandler handler : handlers) - { - boolean ok = handler.latch.await(LATCH_WAIT, TimeUnit.MILLISECONDS); - - assertTrue(ok); - } - - for (ClientSession session : sessions) - { - session.commit(); - } - - sessSend.close(); - for (ClientSession session : sessions) - { - session.close(); - } - - for (int i = 0; i < numSessions; i++) - { - SimpleString subName = new SimpleString(threadNum + "sub" + i); - - s.deleteQueue(subName); - } - - s.close(); - - long end = System.currentTimeMillis(); - - log.info("duration " + (end - start)); - } - - protected void doTestD(final ClientSessionFactory sf, final int threadNum) throws Exception - { - long start = System.currentTimeMillis(); - - ClientSession s = sf.createSession(false, false, false); - - final int numMessages = 100; - - final int numSessions = 10; - - Set<ClientConsumer> consumers = new HashSet<ClientConsumer>(); - Set<ClientSession> sessions = new HashSet<ClientSession>(); - - for (int i = 0; i < numSessions; i++) - { - SimpleString subName = new SimpleString(threadNum + " sub" + i); - - ClientSession sessConsume = sf.createSession(false, false, false); - - sessConsume.createQueue(ADDRESS, subName, null, false); - - ClientConsumer consumer = sessConsume.createConsumer(subName); - - consumers.add(consumer); - - sessions.add(sessConsume); - } - - ClientSession sessSend = sf.createSession(false, false, false); - - ClientProducer producer = sessSend.createProducer(ADDRESS); - - sendMessages(sessSend, producer, numMessages, threadNum); - - sessSend.rollback(); - - sendMessages(sessSend, producer, numMessages, threadNum); - - sessSend.commit(); - - for (ClientSession session : sessions) - { - session.start(); - } - - Set<MyHandler> handlers = new HashSet<MyHandler>(); - - for (ClientConsumer consumer : consumers) - { - MyHandler handler = new MyHandler(threadNum, numMessages); - - consumer.setMessageHandler(handler); - - handlers.add(handler); - } - - for (MyHandler handler : handlers) - { - boolean ok = handler.latch.await(LATCH_WAIT, TimeUnit.MILLISECONDS); - - if (!ok) - { - throw new Exception("Timed out waiting for messages on handler " + System.identityHashCode(handler) + - " threadnum " + - threadNum); - } - - if (handler.failure != null) - { - throw new Exception("Handler failed: " + handler.failure); - } - } - - handlers.clear(); - - // Set handlers to null - for (ClientConsumer consumer : consumers) - { - consumer.setMessageHandler(null); - } - - for (ClientSession session : sessions) - { - session.rollback(); - } - - // New handlers - for (ClientConsumer consumer : consumers) - { - MyHandler handler = new MyHandler(threadNum, numMessages); - - consumer.setMessageHandler(handler); - - handlers.add(handler); - } - - for (MyHandler handler : handlers) - { - boolean ok = handler.latch.await(LATCH_WAIT, TimeUnit.MILLISECONDS); - - if (!ok) - { - throw new Exception("Timed out waiting for messages on handler " + System.identityHashCode(handler) + - " threadnum " + - threadNum); - } - - if (handler.failure != null) - { - throw new Exception("Handler failed on rollback: " + handler.failure); - } - } - - for (ClientSession session : sessions) - { - session.commit(); - } - - sessSend.close(); - for (ClientSession session : sessions) - { - session.close(); - } - - for (int i = 0; i < numSessions; i++) - { - SimpleString subName = new SimpleString(threadNum + " sub" + i); - - s.deleteQueue(subName); - } - - s.close(); - - long end = System.currentTimeMillis(); - - log.info("duration " + (end - start)); - } - - // Now with synchronous receive() - - protected void doTestE(final ClientSessionFactory sf, final int threadNum) throws Exception - { - long start = System.currentTimeMillis(); - - ClientSession s = sf.createSession(false, false, false); - - final int numMessages = 100; - - final int numSessions = 10; - - Set<ClientConsumer> consumers = new HashSet<ClientConsumer>(); - Set<ClientSession> sessions = new HashSet<ClientSession>(); - - for (int i = 0; i < numSessions; i++) - { - SimpleString subName = new SimpleString(threadNum + "sub" + i); - - ClientSession sessConsume = sf.createSession(false, true, true); - - sessConsume.start(); - - sessConsume.createQueue(ADDRESS, subName, null, false); - - ClientConsumer consumer = sessConsume.createConsumer(subName); - - consumers.add(consumer); - - sessions.add(sessConsume); - } - - ClientSession sessSend = sf.createSession(false, true, true); - - ClientProducer producer = sessSend.createProducer(ADDRESS); - - sendMessages(sessSend, producer, numMessages, threadNum); - - consumeMessages(consumers, numMessages, threadNum); - - sessSend.close(); - for (ClientSession session : sessions) - { - session.close(); - } - - for (int i = 0; i < numSessions; i++) - { - SimpleString subName = new SimpleString(threadNum + "sub" + i); - - s.deleteQueue(subName); - } - - s.close(); - - long end = System.currentTimeMillis(); - - log.info("duration " + (end - start)); - } - - protected void doTestF(final ClientSessionFactory sf, final int threadNum) throws Exception - { - long start = System.currentTimeMillis(); - - ClientSession s = sf.createSession(false, false, false); - - final int numMessages = 100; - - final int numSessions = 10; - - Set<ClientConsumer> consumers = new HashSet<ClientConsumer>(); - Set<ClientSession> sessions = new HashSet<ClientSession>(); - - for (int i = 0; i < numSessions; i++) - { - SimpleString subName = new SimpleString(threadNum + "sub" + i); - - ClientSession sessConsume = sf.createSession(false, true, true); - - sessConsume.createQueue(ADDRESS, subName, null, false); - - ClientConsumer consumer = sessConsume.createConsumer(subName); - - consumers.add(consumer); - - sessions.add(sessConsume); - } - - ClientSession sessSend = sf.createSession(false, true, true); - - ClientProducer producer = sessSend.createProducer(ADDRESS); - - sendMessages(sessSend, producer, numMessages, threadNum); - - for (ClientSession session : sessions) - { - session.start(); - } - - consumeMessages(consumers, numMessages, threadNum); - - sessSend.close(); - for (ClientSession session : sessions) - { - session.close(); - } - - for (int i = 0; i < numSessions; i++) - { - SimpleString subName = new SimpleString(threadNum + "sub" + i); - - s.deleteQueue(subName); - } - - s.close(); - - long end = System.currentTimeMillis(); - - log.info("duration " + (end - start)); - } - - protected void doTestG(final ClientSessionFactory sf, final int threadNum) throws Exception - { - long start = System.currentTimeMillis(); - - ClientSession s = sf.createSession(false, false, false); - - final int numMessages = 100; - - final int numSessions = 10; - - Set<ClientConsumer> consumers = new HashSet<ClientConsumer>(); - Set<ClientSession> sessions = new HashSet<ClientSession>(); - - for (int i = 0; i < numSessions; i++) - { - SimpleString subName = new SimpleString(threadNum + "sub" + i); - - ClientSession sessConsume = sf.createSession(false, false, false); - - sessConsume.start(); - - sessConsume.createQueue(ADDRESS, subName, null, false); - - ClientConsumer consumer = sessConsume.createConsumer(subName); - - consumers.add(consumer); - - sessions.add(sessConsume); - } - - ClientSession sessSend = sf.createSession(false, false, false); - - ClientProducer producer = sessSend.createProducer(ADDRESS); - - sendMessages(sessSend, producer, numMessages, threadNum); - - sessSend.rollback(); - - sendMessages(sessSend, producer, numMessages, threadNum); - - sessSend.commit(); - - consumeMessages(consumers, numMessages, threadNum); - - for (ClientSession session : sessions) - { - session.rollback(); - } - - consumeMessages(consumers, numMessages, threadNum); - - for (ClientSession session : sessions) - { - session.commit(); - } - - sessSend.close(); - for (ClientSession session : sessions) - { - session.close(); - } - - for (int i = 0; i < numSessions; i++) - { - SimpleString subName = new SimpleString(threadNum + "sub" + i); - - s.deleteQueue(subName); - } - - s.close(); - - long end = System.currentTimeMillis(); - - log.info("duration " + (end - start)); - } - - protected void doTestH(final ClientSessionFactory sf, final int threadNum) throws Exception - { - long start = System.currentTimeMillis(); - - ClientSession s = sf.createSession(false, false, false); - - final int numMessages = 100; - - final int numSessions = 10; - - Set<ClientConsumer> consumers = new HashSet<ClientConsumer>(); - Set<ClientSession> sessions = new HashSet<ClientSession>(); - - for (int i = 0; i < numSessions; i++) - { - SimpleString subName = new SimpleString(threadNum + "sub" + i); - - ClientSession sessConsume = sf.createSession(false, false, false); - - sessConsume.createQueue(ADDRESS, subName, null, false); - - ClientConsumer consumer = sessConsume.createConsumer(subName); - - consumers.add(consumer); - - sessions.add(sessConsume); - } - - ClientSession sessSend = sf.createSession(false, false, false); - - ClientProducer producer = sessSend.createProducer(ADDRESS); - - sendMessages(sessSend, producer, numMessages, threadNum); - - sessSend.rollback(); - - sendMessages(sessSend, producer, numMessages, threadNum); - - sessSend.commit(); - - for (ClientSession session : sessions) - { - session.start(); - } - - consumeMessages(consumers, numMessages, threadNum); - - for (ClientSession session : sessions) - { - session.rollback(); - } - - consumeMessages(consumers, numMessages, threadNum); - - for (ClientSession session : sessions) - { - session.commit(); - } - - sessSend.close(); - for (ClientSession session : sessions) - { - session.close(); - } - - for (int i = 0; i < numSessions; i++) - { - SimpleString subName = new SimpleString(threadNum + "sub" + i); - - s.deleteQueue(subName); - } - - s.close(); - - long end = System.currentTimeMillis(); - - log.info("duration " + (end - start)); - } - - protected void doTestI(final ClientSessionFactory sf, final int threadNum) throws Exception - { - ClientSession sessCreate = sf.createSession(false, true, true); - - sessCreate.createQueue(ADDRESS, new SimpleString(threadNum + ADDRESS.toString()), null, false); - - ClientSession sess = sf.createSession(false, true, true); - - sess.start(); - - ClientConsumer consumer = sess.createConsumer(new SimpleString(threadNum + ADDRESS.toString())); - - ClientProducer producer = sess.createProducer(ADDRESS); - - ClientMessage message = sess.createClientMessage(HornetQTextMessage.TYPE, - false, - 0, - System.currentTimeMillis(), - (byte)1); - producer.send(message); - - ClientMessage message2 = consumer.receive(RECEIVE_TIMEOUT); - - assertNotNull(message2); - - message2.acknowledge(); - - sess.close(); - - sessCreate.deleteQueue(new SimpleString(threadNum + ADDRESS.toString())); - - sessCreate.close(); - } - - protected void doTestJ(final ClientSessionFactory sf, final int threadNum) throws Exception - { - ClientSession sessCreate = sf.createSession(false, true, true); - - sessCreate.createQueue(ADDRESS, new SimpleString(threadNum + ADDRESS.toString()), null, false); - - ClientSession sess = sf.createSession(false, true, true); - - sess.start(); - - ClientConsumer consumer = sess.createConsumer(new SimpleString(threadNum + ADDRESS.toString())); - - ClientProducer producer = sess.createProducer(ADDRESS); - - ClientMessage message = sess.createClientMessage(HornetQTextMessage.TYPE, - false, - 0, - System.currentTimeMillis(), - (byte)1); - producer.send(message); - - ClientMessage message2 = consumer.receive(RECEIVE_TIMEOUT); - - assertNotNull(message2); - - message2.acknowledge(); - - sess.close(); - - sessCreate.deleteQueue(new SimpleString(threadNum + ADDRESS.toString())); - - sessCreate.close(); - } - - protected void doTestK(final ClientSessionFactory sf, final int threadNum) throws Exception - { - ClientSession s = sf.createSession(false, false, false); - - s.createQueue(ADDRESS, new SimpleString(threadNum + ADDRESS.toString()), null, false); - - final int numConsumers = 100; - - for (int i = 0; i < numConsumers; i++) - { - ClientConsumer consumer = s.createConsumer(new SimpleString(threadNum + ADDRESS.toString())); - - consumer.close(); - } - - s.deleteQueue(new SimpleString(threadNum + ADDRESS.toString())); - - s.close(); - } - - /* - * This test tests failure during create connection - */ - protected void doTestL(final ClientSessionFactory sf) throws Exception - { - final int numSessions = 10; - - for (int i = 0; i < numSessions; i++) - { - ClientSession session = sf.createSession(false, false, false); - - session.close(); - } - } - - // Browsers - // FIXME - this test won't work until we use a proper iterator for browsing a queue. - // Making a copy of the queue for a browser consumer doesn't work well with replication since - // When replicating the create consumer (browser) to the backup, when executed on the backup the - // backup may have different messages in its queue since been added on different threads. - // So when replicating deliveries they may not be found. - // https://jira.jboss.org/jira/browse/JBMESSAGING-1433 - // protected void doTestM(final ClientSessionFactory sf, final int threadNum) throws Exception - // { - // long start = System.currentTimeMillis(); - // - // ClientSession sessSend = sf.createSession(false, true, true, false); - // - // ClientSession sessConsume = sf.createSession(false, true, true, false); - // - // sessConsume.createQueue(ADDRESS, new SimpleString(threadNum + "sub"), null, false, false); - // - // final int numMessages = 100; - // - // ClientProducer producer = sessSend.createProducer(ADDRESS); - // - // sendMessages(sessSend, producer, numMessages, threadNum); - // - // ClientConsumer browser = sessConsume.createConsumer(new SimpleString(threadNum + "sub"), - // null, false, true); - // - // Map<Integer, Integer> consumerCounts = new HashMap<Integer, Integer>(); - // - // for (int i = 0; i < numMessages; i++) - // { - // ClientMessage msg = browser.receive(RECEIVE_TIMEOUT); - // - // assertNotNull(msg); - // - // int tn = (Integer)msg.getProperty(new SimpleString("threadnum")); - // int cnt = (Integer)msg.getProperty(new SimpleString("count")); - // - // Integer c = consumerCounts.get(tn); - // if (c == null) - // { - // c = new Integer(cnt); - // } - // - // if (cnt != c.intValue()) - // { - // throw new Exception("Invalid count, expected " + c + " got " + cnt); - // } - // - // c++; - // - // //Wrap - // if (c == numMessages) - // { - // c = 0; - // } - // - // consumerCounts.put(tn, c); - // - // msg.acknowledge(); - // } - // - // sessConsume.close(); - // - // sessConsume = sf.createSession(false, true, true, false); - // - // browser = sessConsume.createConsumer(new SimpleString(threadNum + "sub"), - // null, false, true); - // - // //Messages should still be there - // - // consumerCounts.clear(); - // - // for (int i = 0; i < numMessages; i++) - // { - // ClientMessage msg = browser.receive(RECEIVE_TIMEOUT); - // - // assertNotNull(msg); - // - // int tn = (Integer)msg.getProperty(new SimpleString("threadnum")); - // int cnt = (Integer)msg.getProperty(new SimpleString("count")); - // - // Integer c = consumerCounts.get(tn); - // if (c == null) - // { - // c = new Integer(cnt); - // } - // - // if (cnt != c.intValue()) - // { - // throw new Exception("Invalid count, expected " + c + " got " + cnt); - // } - // - // c++; - // - // //Wrap - // if (c == numMessages) - // { - // c = 0; - // } - // - // consumerCounts.put(tn, c); - // - // msg.acknowledge(); - // } - // - // sessConsume.close(); - // - // sessSend.deleteQueue(new SimpleString(threadNum + "sub")); - // - // sessSend.close(); - // - // long end = System.currentTimeMillis(); - // - // log.info("duration " + (end - start)); - // } - - protected void doTestN(final ClientSessionFactory sf, final int threadNum) throws Exception - { - ClientSession sessCreate = sf.createSession(false, true, true); - - sessCreate.createQueue(ADDRESS, new SimpleString(threadNum + ADDRESS.toString()), null, false); - - ClientSession sess = sf.createSession(false, true, true); - - sess.stop(); - - sess.start(); - - sess.stop(); - - ClientConsumer consumer = sess.createConsumer(new SimpleString(threadNum + ADDRESS.toString())); - - ClientProducer producer = sess.createProducer(ADDRESS); - - ClientMessage message = sess.createClientMessage(HornetQTextMessage.TYPE, - false, - 0, - System.currentTimeMillis(), - (byte)1); - producer.send(message); - - sess.start(); - - ClientMessage message2 = consumer.receive(RECEIVE_TIMEOUT); - - assertNotNull(message2); - - message2.acknowledge(); - - sess.stop(); - - sess.start(); - - sess.close(); - - sessCreate.deleteQueue(new SimpleString(threadNum + ADDRESS.toString())); - - sessCreate.close(); - } - - protected int getLatchWait() - { - return 60000; - } - - protected int getNumIterations() - { - return 3; - } - - @Override - protected void setUp() throws Exception - { - super.setUp(); - - log.info("************ Starting test " + getName()); - } - - @Override - protected void tearDown() throws Exception - { - if (liveServer != null && liveServer.isStarted()) - { - liveServer.stop(); - } - - liveServer = null; - - super.tearDown(); - } - - // Private ------------------------------------------------------- - - private void runTestMultipleThreads(final RunnableT runnable, - final int numThreads, - final boolean failOnCreateConnection) throws Exception - { - runTestMultipleThreads(runnable, numThreads, failOnCreateConnection, 1000); - } - - private void runTestMultipleThreads(final RunnableT runnable, - final int numThreads, - final boolean failOnCreateConnection, - final long failDelay) throws Exception - { - - runMultipleThreadsFailoverTest(runnable, numThreads, getNumIterations(), failOnCreateConnection, failDelay); - } - - /** - * @return - */ - protected ClientSessionFactoryInternal createSessionFactory() - { - final ClientSessionFactoryInternal sf = new ClientSessionFactoryImpl(new TransportConfiguration("org.hornetq.core.remoting.impl.invm.InVMConnectorFactory")); - sf.setReconnectAttempts(-1); - sf.setProducerWindowSize(32 * 1024); - - return sf; - } - - protected void stop() throws Exception - { - liveServer.stop(); - - System.gc(); - - assertEquals(0, InVMRegistry.instance.size()); - } - - private void sendMessages(final ClientSession sessSend, - final ClientProducer producer, - final int numMessages, - final int threadNum) throws Exception - { - for (int i = 0; i < numMessages; i++) - { - ClientMessage message = sessSend.createClientMessage(HornetQBytesMessage.TYPE, - false, - 0, - System.currentTimeMillis(), - (byte)1); - message.putIntProperty(new SimpleString("threadnum"), threadNum); - message.putIntProperty(new SimpleString("count"), i); - setBody(message); - producer.send(message); - } - } - - private void consumeMessages(final Set<ClientConsumer> consumers, final int numMessages, final int threadNum) throws Exception - { - // We make sure the messages arrive in the order they were sent from a particular producer - Map<ClientConsumer, Map<Integer, Integer>> counts = new HashMap<ClientConsumer, Map<Integer, Integer>>(); - - for (int i = 0; i < numMessages; i++) - { - for (ClientConsumer consumer : consumers) - { - Map<Integer, Integer> consumerCounts = counts.get(consumer); - - if (consumerCounts == null) - { - consumerCounts = new HashMap<Integer, Integer>(); - counts.put(consumer, consumerCounts); - } - - ClientMessage msg = consumer.receive(RECEIVE_TIMEOUT); - - assertNotNull(msg); - - int tn = (Integer)msg.getProperty(new SimpleString("threadnum")); - int cnt = (Integer)msg.getProperty(new SimpleString("count")); - - Integer c = consumerCounts.get(tn); - if (c == null) - { - c = new Integer(cnt); - } - - if (tn == threadNum && cnt != c.intValue()) - { - throw new Exception("Invalid count, expected " + tn + ": " + c + " got " + cnt); - } - - c++; - - // Wrap - if (c == numMessages) - { - c = 0; - } - - consumerCounts.put(tn, c); - - msg.acknowledge(); - } - } - } - - // Inner classes ------------------------------------------------- - - private class MyHandler implements MessageHandler - { - CountDownLatch latch = new CountDownLatch(1); - - private final Map<Integer, Integer> counts = new HashMap<Integer, Integer>(); - - volatile String failure; - - final int tn; - - final int numMessages; - - volatile boolean done; - - synchronized void reset() - { - counts.clear(); - - done = false; - - failure = null; - - latch = new CountDownLatch(1); - } - - MyHandler(final int threadNum, final int numMessages) - { - tn = threadNum; - - this.numMessages = numMessages; - } - - public synchronized void onMessage(final ClientMessage message) - { - try - { - message.acknowledge(); - } - catch (HornetQException me) - { - log.error("Failed to process", me); - } - - if (done) - { - return; - } - - int threadNum = (Integer)message.getProperty(new SimpleString("threadnum")); - int cnt = (Integer)message.getProperty(new SimpleString("count")); - - Integer c = counts.get(threadNum); - if (c == null) - { - c = new Integer(cnt); - } - - if (tn == threadNum && cnt != c.intValue()) - { - failure = "Invalid count, expected " + threadNum + ":" + c + " got " + cnt; - log.error(failure); - - latch.countDown(); - } - - if (!checkSize(message)) - { - failure = "Invalid size on message"; - log.error(failure); - latch.countDown(); - } - - if (tn == threadNum && c == numMessages - 1) - { - done = true; - latch.countDown(); - } - - c++; - // Wrap around at numMessages - if (c == numMessages) - { - c = 0; - } - - counts.put(threadNum, c); - - } - } -} \ No newline at end of file Added: branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/integration/cluster/failover/NettyAsynchronousFailoverTest.java =================================================================== --- branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/integration/cluster/failover/NettyAsynchronousFailoverTest.java (rev 0) +++ branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/integration/cluster/failover/NettyAsynchronousFailoverTest.java 2009-09-22 15:08:41 UTC (rev 7979) @@ -0,0 +1,64 @@ +/* + * Copyright 2009 Red Hat, Inc. + * Red Hat licenses this file to you under the Apache License, version + * 2.0 (the "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * http://www.apache.org/licenses/LICENSE-2.0 + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or + * implied. See the License for the specific language governing + * permissions and limitations under the License. + */ + +package org.hornetq.tests.integration.cluster.failover; + +import java.util.HashMap; +import java.util.Map; + +import org.hornetq.core.config.TransportConfiguration; +import org.hornetq.integration.transports.netty.TransportConstants; + +/** + * A NettyAsynchronousFailoverTest + * + * @author tim + * + * + */ +public class NettyAsynchronousFailoverTest extends AsynchronousFailoverTest +{ + @Override + protected TransportConfiguration getAcceptorTransportConfiguration(boolean live) + { + if (live) + { + return new TransportConfiguration("org.hornetq.integration.transports.netty.NettyAcceptorFactory"); + } + else + { + Map<String, Object> server1Params = new HashMap<String, Object>(); + + server1Params.put(TransportConstants.PORT_PROP_NAME, TransportConstants.DEFAULT_PORT + 1); + + return new TransportConfiguration("org.hornetq.integration.transports.netty.NettyAcceptorFactory", server1Params); + } + } + + @Override + protected TransportConfiguration getConnectorTransportConfiguration(final boolean live) + { + if (live) + { + return new TransportConfiguration("org.hornetq.integration.transports.netty.NettyConnectorFactory"); + } + else + { + Map<String, Object> server1Params = new HashMap<String, Object>(); + + server1Params.put(TransportConstants.PORT_PROP_NAME, TransportConstants.DEFAULT_PORT + 1); + + return new TransportConfiguration("org.hornetq.integration.transports.netty.NettyConnectorFactory", server1Params); + } + } +} Added: branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/integration/cluster/failover/NettyFailoverTest.java =================================================================== --- branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/integration/cluster/failover/NettyFailoverTest.java (rev 0) +++ branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/integration/cluster/failover/NettyFailoverTest.java 2009-09-22 15:08:41 UTC (rev 7979) @@ -0,0 +1,84 @@ +/* + * Copyright 2009 Red Hat, Inc. + * Red Hat licenses this file to you under the Apache License, version + * 2.0 (the "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * http://www.apache.org/licenses/LICENSE-2.0 + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or + * implied. See the License for the specific language governing + * permissions and limitations under the License. + */ + +package org.hornetq.tests.integration.cluster.failover; + +import java.util.HashMap; +import java.util.Map; + +import org.hornetq.core.config.TransportConfiguration; +import org.hornetq.integration.transports.netty.TransportConstants; + +/** + * A NettyFailoverTest + * + * @author <a href="mailto:tim.fox@jboss.com">Tim Fox</a> + * + * + */ +public class NettyFailoverTest extends FailoverTest +{ + + // Constants ----------------------------------------------------- + + // Attributes ---------------------------------------------------- + + // Static -------------------------------------------------------- + + // Constructors -------------------------------------------------- + + // Public -------------------------------------------------------- + + // Package protected --------------------------------------------- + + // Protected ----------------------------------------------------- + + @Override + protected TransportConfiguration getAcceptorTransportConfiguration(boolean live) + { + if (live) + { + return new TransportConfiguration("org.hornetq.integration.transports.netty.NettyAcceptorFactory"); + } + else + { + Map<String, Object> server1Params = new HashMap<String, Object>(); + + server1Params.put(TransportConstants.PORT_PROP_NAME, TransportConstants.DEFAULT_PORT + 1); + + return new TransportConfiguration("org.hornetq.integration.transports.netty.NettyAcceptorFactory", server1Params); + } + } + + @Override + protected TransportConfiguration getConnectorTransportConfiguration(final boolean live) + { + if (live) + { + return new TransportConfiguration("org.hornetq.integration.transports.netty.NettyConnectorFactory"); + } + else + { + Map<String, Object> server1Params = new HashMap<String, Object>(); + + server1Params.put(TransportConstants.PORT_PROP_NAME, TransportConstants.DEFAULT_PORT + 1); + + return new TransportConfiguration("org.hornetq.integration.transports.netty.NettyConnectorFactory", server1Params); + } + } + + // Private ------------------------------------------------------- + + // Inner classes ------------------------------------------------- + +} Deleted: branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/integration/cluster/failover/NettyMultiThreadRandomReattachTest.java =================================================================== --- branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/integration/cluster/failover/NettyMultiThreadRandomReattachTest.java 2009-09-22 13:57:55 UTC (rev 7978) +++ branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/integration/cluster/failover/NettyMultiThreadRandomReattachTest.java 2009-09-22 15:08:41 UTC (rev 7979) @@ -1,64 +0,0 @@ -/* - * Copyright 2009 Red Hat, Inc. - * Red Hat licenses this file to you under the Apache License, version - * 2.0 (the "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * http://www.apache.org/licenses/LICENSE-2.0 - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or - * implied. See the License for the specific language governing - * permissions and limitations under the License. - */ - -package org.hornetq.tests.integration.cluster.failover; - -import java.util.HashMap; -import java.util.Map; - -import org.hornetq.core.client.impl.ClientSessionFactoryImpl; -import org.hornetq.core.client.impl.ClientSessionFactoryInternal; -import org.hornetq.core.config.Configuration; -import org.hornetq.core.config.TransportConfiguration; -import org.hornetq.core.config.impl.ConfigurationImpl; -import org.hornetq.core.server.HornetQ; - -/** - * A NettyMultiThreadRandomReattachTest - * - * @author <a href="mailto:tim.fox@jboss.com">Tim Fox</a> - * - * Created 18 Feb 2009 08:01:20 - * - * - */ -public class NettyMultiThreadRandomReattachTest extends MultiThreadRandomReattachTest -{ - @Override - protected void start() throws Exception - { - Configuration liveConf = new ConfigurationImpl(); - liveConf.setJMXManagementEnabled(false); - liveConf.setSecurityEnabled(false); - liveConf.getAcceptorConfigurations().clear(); - liveConf.getAcceptorConfigurations() - .add(new TransportConfiguration("org.hornetq.integration.transports.netty.NettyAcceptorFactory")); - Map<String, TransportConfiguration> connectors = new HashMap<String, TransportConfiguration>(); - TransportConfiguration backupTC = new TransportConfiguration("org.hornetq.integration.transports.netty.NettyConnectorFactory"); - connectors.put(backupTC.getName(), backupTC); - liveConf.setConnectorConfigurations(connectors); - liveConf.setBackupConnectorName(backupTC.getName()); - liveServer = HornetQ.newHornetQServer(liveConf, false); - liveServer.start(); - } - - @Override - protected ClientSessionFactoryInternal createSessionFactory() - { - final ClientSessionFactoryInternal sf = new ClientSessionFactoryImpl(new TransportConfiguration("org.hornetq.integration.transports.netty.NettyConnectorFactory")); - - sf.setProducerWindowSize(32 * 1024); - return sf; - } - -} Deleted: branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/integration/cluster/failover/RandomReattachTest.java =================================================================== --- branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/integration/cluster/failover/RandomReattachTest.java 2009-09-22 13:57:55 UTC (rev 7978) +++ branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/integration/cluster/failover/RandomReattachTest.java 2009-09-22 15:08:41 UTC (rev 7979) @@ -1,1486 +0,0 @@ -/* - * Copyright 2009 Red Hat, Inc. - * Red Hat licenses this file to you under the Apache License, version - * 2.0 (the "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * http://www.apache.org/licenses/LICENSE-2.0 - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or - * implied. See the License for the specific language governing - * permissions and limitations under the License. - */ - -package org.hornetq.tests.integration.cluster.failover; - -import java.util.HashMap; -import java.util.HashSet; -import java.util.Map; -import java.util.Set; -import java.util.Timer; -import java.util.TimerTask; -import java.util.concurrent.CountDownLatch; -import java.util.concurrent.TimeUnit; - -import org.hornetq.core.client.ClientConsumer; -import org.hornetq.core.client.ClientMessage; -import org.hornetq.core.client.ClientProducer; -import org.hornetq.core.client.ClientSession; -import org.hornetq.core.client.ClientSessionFactory; -import org.hornetq.core.client.MessageHandler; -import org.hornetq.core.client.impl.ClientSessionFactoryImpl; -import org.hornetq.core.client.impl.ClientSessionInternal; -import org.hornetq.core.config.Configuration; -import org.hornetq.core.config.TransportConfiguration; -import org.hornetq.core.config.impl.ConfigurationImpl; -import org.hornetq.core.exception.HornetQException; -import org.hornetq.core.logging.Logger; -import org.hornetq.core.remoting.impl.invm.InVMRegistry; -import org.hornetq.core.server.HornetQ; -import org.hornetq.core.server.HornetQServer; -import org.hornetq.jms.client.HornetQTextMessage; -import org.hornetq.tests.util.UnitTestCase; -import org.hornetq.utils.SimpleString; - -/** - * A RandomFailoverSoakTest - * - * @author <a href="mailto:tim.fox@jboss.com">Tim Fox</a> - */ -public class RandomReattachTest extends UnitTestCase -{ - private static final Logger log = Logger.getLogger(RandomReattachTest.class); - - // Constants ----------------------------------------------------- - - private static final int RECEIVE_TIMEOUT = 10000; - - // Attributes ---------------------------------------------------- - - private static final SimpleString ADDRESS = new SimpleString("FailoverTestAddress"); - - private HornetQServer liveService; - - private Timer timer; - - // Static -------------------------------------------------------- - - // Constructors -------------------------------------------------- - - // Public -------------------------------------------------------- - - public void testA() throws Exception - { - runTest(new RunnableT() - { - public void run(final ClientSessionFactory sf) throws Exception - { - doTestA(sf); - } - }); - } - - public void testB() throws Exception - { - runTest(new RunnableT() - { - public void run(final ClientSessionFactory sf) throws Exception - { - doTestB(sf); - } - }); - } - - public void testC() throws Exception - { - runTest(new RunnableT() - { - public void run(final ClientSessionFactory sf) throws Exception - { - doTestC(sf); - } - }); - } - - public void testD() throws Exception - { - runTest(new RunnableT() - { - public void run(final ClientSessionFactory sf) throws Exception - { - doTestD(sf); - } - }); - } - - public void testE() throws Exception - { - runTest(new RunnableT() - { - public void run(final ClientSessionFactory sf) throws Exception - { - doTestE(sf); - } - }); - } - - public void testF() throws Exception - { - runTest(new RunnableT() - { - public void run(final ClientSessionFactory sf) throws Exception - { - doTestF(sf); - } - }); - } - - public void testG() throws Exception - { - runTest(new RunnableT() - { - public void run(final ClientSessionFactory sf) throws Exception - { - doTestG(sf); - } - }); - } - - public void testH() throws Exception - { - runTest(new RunnableT() - { - public void run(final ClientSessionFactory sf) throws Exception - { - doTestH(sf); - } - }); - } - - public void testI() throws Exception - { - runTest(new RunnableT() - { - public void run(final ClientSessionFactory sf) throws Exception - { - doTestI(sf); - } - }); - } - - public void testJ() throws Exception - { - runTest(new RunnableT() - { - public void run(final ClientSessionFactory sf) throws Exception - { - doTestJ(sf); - } - }); - } - - public void testK() throws Exception - { - runTest(new RunnableT() - { - public void run(final ClientSessionFactory sf) throws Exception - { - doTestK(sf); - } - }); - } - - public void testL() throws Exception - { - runTest(new RunnableT() - { - public void run(final ClientSessionFactory sf) throws Exception - { - doTestL(sf); - } - }); - } - - public void testN() throws Exception - { - runTest(new RunnableT() - { - public void run(final ClientSessionFactory sf) throws Exception - { - doTestN(sf); - } - }); - } - - public void runTest(final RunnableT runnable) throws Exception - { - final int numIts = getNumIterations(); - - for (int its = 0; its < numIts; its++) - { - start(); - - ClientSessionFactoryImpl sf = new ClientSessionFactoryImpl(new TransportConfiguration("org.hornetq.core.remoting.impl.invm.InVMConnectorFactory")); - - sf.setProducerWindowSize(32 * 1024); - - ClientSession session = sf.createSession(false, false, false); - - Failer failer = startFailer(1000, session); - - do - { - runnable.run(sf); - } - while (!failer.isExecuted()); - - session.close(); - - assertEquals(0, sf.numSessions()); - - assertEquals(0, sf.numConnections()); - - stop(); - } - } - - // Package protected --------------------------------------------- - - // Protected ----------------------------------------------------- - - protected void doTestA(final ClientSessionFactory sf) throws Exception - { - long start = System.currentTimeMillis(); - - ClientSession s = sf.createSession(false, false, false); - - final int numMessages = 100; - - final int numSessions = 10; - - Set<ClientConsumer> consumers = new HashSet<ClientConsumer>(); - Set<ClientSession> sessions = new HashSet<ClientSession>(); - - for (int i = 0; i < numSessions; i++) - { - SimpleString subName = new SimpleString("sub" + i); - - ClientSession sessConsume = sf.createSession(false, true, true); - - sessConsume.start(); - - sessConsume.createQueue(ADDRESS, subName, null, false); - - ClientConsumer consumer = sessConsume.createConsumer(subName); - - consumers.add(consumer); - - sessions.add(sessConsume); - } - - ClientSession sessSend = sf.createSession(false, true, true); - - ClientProducer producer = sessSend.createProducer(ADDRESS); - - for (int i = 0; i < numMessages; i++) - { - ClientMessage message = sessSend.createClientMessage(HornetQTextMessage.TYPE, - false, - 0, - System.currentTimeMillis(), - (byte)1); - message.putIntProperty(new SimpleString("count"), i); - producer.send(message); - } - - class MyHandler implements MessageHandler - { - final CountDownLatch latch = new CountDownLatch(1); - - volatile int count; - - public void onMessage(ClientMessage message) - { - if (count == numMessages) - { - fail("Too many messages"); - } - - assertEquals(count, message.getProperty(new SimpleString("count"))); - - count++; - - try - { - message.acknowledge(); - } - catch (HornetQException me) - { - log.error("Failed to process", me); - } - - if (count == numMessages) - { - latch.countDown(); - } - } - } - - Set<MyHandler> handlers = new HashSet<MyHandler>(); - - for (ClientConsumer consumer : consumers) - { - MyHandler handler = new MyHandler(); - - consumer.setMessageHandler(handler); - - handlers.add(handler); - } - - for (MyHandler handler : handlers) - { - boolean ok = handler.latch.await(5000, TimeUnit.MILLISECONDS); - - assertTrue("Didn't receive all messages", ok); - } - - sessSend.close(); - for (ClientSession session : sessions) - { - session.close(); - } - - for (int i = 0; i < numSessions; i++) - { - SimpleString subName = new SimpleString("sub" + i); - - s.deleteQueue(subName); - } - - s.close(); - - long end = System.currentTimeMillis(); - - log.info("duration " + (end - start)); - } - - protected void doTestB(final ClientSessionFactory sf) throws Exception - { - long start = System.currentTimeMillis(); - - ClientSession s = sf.createSession(false, false, false); - - final int numMessages = 100; - - final int numSessions = 50; - - Set<ClientConsumer> consumers = new HashSet<ClientConsumer>(); - Set<ClientSession> sessions = new HashSet<ClientSession>(); - - for (int i = 0; i < numSessions; i++) - { - SimpleString subName = new SimpleString("sub" + i); - - ClientSession sessConsume = sf.createSession(false, true, true); - - sessConsume.createQueue(ADDRESS, subName, null, false); - - ClientConsumer consumer = sessConsume.createConsumer(subName); - - consumers.add(consumer); - - sessions.add(sessConsume); - } - - ClientSession sessSend = sf.createSession(false, true, true); - - ClientProducer producer = sessSend.createProducer(ADDRESS); - - for (int i = 0; i < numMessages; i++) - { - ClientMessage message = sessSend.createClientMessage(HornetQTextMessage.TYPE, - false, - 0, - System.currentTimeMillis(), - (byte)1); - message.putIntProperty(new SimpleString("count"), i); - producer.send(message); - } - - for (ClientSession session : sessions) - { - session.start(); - } - - class MyHandler implements MessageHandler - { - final CountDownLatch latch = new CountDownLatch(1); - - volatile int count; - - public void onMessage(ClientMessage message) - { - if (count == numMessages) - { - fail("Too many messages"); - } - - assertEquals(count, message.getProperty(new SimpleString("count"))); - - count++; - - if (count == numMessages) - { - latch.countDown(); - } - } - } - - Set<MyHandler> handlers = new HashSet<MyHandler>(); - - for (ClientConsumer consumer : consumers) - { - MyHandler handler = new MyHandler(); - - consumer.setMessageHandler(handler); - - handlers.add(handler); - } - - for (MyHandler handler : handlers) - { - boolean ok = handler.latch.await(10000, TimeUnit.MILLISECONDS); - - assertTrue(ok); - } - - sessSend.close(); - - for (ClientSession session : sessions) - { - session.close(); - } - - for (int i = 0; i < numSessions; i++) - { - SimpleString subName = new SimpleString("sub" + i); - - s.deleteQueue(subName); - } - - s.close(); - - long end = System.currentTimeMillis(); - - log.info("duration " + (end - start)); - - } - - protected void doTestC(final ClientSessionFactory sf) throws Exception - { - long start = System.currentTimeMillis(); - - ClientSession s = sf.createSession(false, false, false); - - final int numMessages = 100; - - final int numSessions = 10; - - Set<ClientConsumer> consumers = new HashSet<ClientConsumer>(); - Set<ClientSession> sessions = new HashSet<ClientSession>(); - - for (int i = 0; i < numSessions; i++) - { - SimpleString subName = new SimpleString("sub" + i); - - ClientSession sessConsume = sf.createSession(false, false, false); - - sessConsume.start(); - - sessConsume.createQueue(ADDRESS, subName, null, false); - - ClientConsumer consumer = sessConsume.createConsumer(subName); - - consumers.add(consumer); - - sessions.add(sessConsume); - } - - ClientSession sessSend = sf.createSession(false, true, true); - - ClientProducer producer = sessSend.createProducer(ADDRESS); - - for (int i = 0; i < numMessages; i++) - { - ClientMessage message = sessSend.createClientMessage(HornetQTextMessage.TYPE, - false, - 0, - System.currentTimeMillis(), - (byte)1); - message.putIntProperty(new SimpleString("count"), i); - producer.send(message); - } - - sessSend.rollback(); - - for (int i = 0; i < numMessages; i++) - { - ClientMessage message = sessSend.createClientMessage(HornetQTextMessage.TYPE, - false, - 0, - System.currentTimeMillis(), - (byte)1); - message.putIntProperty(new SimpleString("count"), i); - producer.send(message); - } - - sessSend.commit(); - - class MyHandler implements MessageHandler - { - final CountDownLatch latch = new CountDownLatch(1); - - volatile int count; - - public void onMessage(ClientMessage message) - { - if (count == numMessages) - { - fail("Too many messages"); - } - - assertEquals(count, message.getProperty(new SimpleString("count"))); - - count++; - - if (count == numMessages) - { - latch.countDown(); - } - } - } - - Set<MyHandler> handlers = new HashSet<MyHandler>(); - - for (ClientConsumer consumer : consumers) - { - MyHandler handler = new MyHandler(); - - consumer.setMessageHandler(handler); - - handlers.add(handler); - } - - for (MyHandler handler : handlers) - { - boolean ok = handler.latch.await(10000, TimeUnit.MILLISECONDS); - - assertTrue(ok); - } - - handlers.clear(); - - // New handlers - for (ClientConsumer consumer : consumers) - { - MyHandler handler = new MyHandler(); - - consumer.setMessageHandler(handler); - - handlers.add(handler); - } - - for (ClientSession session : sessions) - { - session.rollback(); - } - - for (MyHandler handler : handlers) - { - boolean ok = handler.latch.await(10000, TimeUnit.MILLISECONDS); - - assertTrue(ok); - } - - for (ClientSession session : sessions) - { - session.commit(); - } - - sessSend.close(); - for (ClientSession session : sessions) - { - session.close(); - } - - for (int i = 0; i < numSessions; i++) - { - SimpleString subName = new SimpleString("sub" + i); - - s.deleteQueue(subName); - } - - s.close(); - - long end = System.currentTimeMillis(); - - log.info("duration " + (end - start)); - } - - protected void doTestD(final ClientSessionFactory sf) throws Exception - { - long start = System.currentTimeMillis(); - - ClientSession s = sf.createSession(false, false, false); - - final int numMessages = 100; - - final int numSessions = 10; - - Set<ClientConsumer> consumers = new HashSet<ClientConsumer>(); - Set<ClientSession> sessions = new HashSet<ClientSession>(); - - for (int i = 0; i < numSessions; i++) - { - SimpleString subName = new SimpleString("sub" + i); - - ClientSession sessConsume = sf.createSession(false, false, false); - - sessConsume.createQueue(ADDRESS, subName, null, false); - - ClientConsumer consumer = sessConsume.createConsumer(subName); - - consumers.add(consumer); - - sessions.add(sessConsume); - } - - ClientSession sessSend = sf.createSession(false, true, true); - - ClientProducer producer = sessSend.createProducer(ADDRESS); - - for (int i = 0; i < numMessages; i++) - { - ClientMessage message = sessSend.createClientMessage(HornetQTextMessage.TYPE, - false, - 0, - System.currentTimeMillis(), - (byte)1); - message.putIntProperty(new SimpleString("count"), i); - producer.send(message); - } - - sessSend.rollback(); - - for (int i = 0; i < numMessages; i++) - { - ClientMessage message = sessSend.createClientMessage(HornetQTextMessage.TYPE, - false, - 0, - System.currentTimeMillis(), - (byte)1); - message.putIntProperty(new SimpleString("count"), i); - producer.send(message); - } - - sessSend.commit(); - - for (ClientSession session : sessions) - { - session.start(); - } - - class MyHandler implements MessageHandler - { - final CountDownLatch latch = new CountDownLatch(1); - - volatile int count; - - public void onMessage(ClientMessage message) - { - if (count == numMessages) - { - fail("Too many messages"); - } - - assertEquals(count, message.getProperty(new SimpleString("count"))); - - count++; - - if (count == numMessages) - { - latch.countDown(); - } - } - } - - Set<MyHandler> handlers = new HashSet<MyHandler>(); - - for (ClientConsumer consumer : consumers) - { - MyHandler handler = new MyHandler(); - - consumer.setMessageHandler(handler); - - handlers.add(handler); - } - - for (MyHandler handler : handlers) - { - boolean ok = handler.latch.await(10000, TimeUnit.MILLISECONDS); - - assertTrue(ok); - } - - handlers.clear(); - - // New handlers - for (ClientConsumer consumer : consumers) - { - MyHandler handler = new MyHandler(); - - consumer.setMessageHandler(handler); - - handlers.add(handler); - } - - for (ClientSession session : sessions) - { - session.rollback(); - } - - for (MyHandler handler : handlers) - { - boolean ok = handler.latch.await(10000, TimeUnit.MILLISECONDS); - - assertTrue(ok); - } - - for (ClientSession session : sessions) - { - session.commit(); - } - - sessSend.close(); - for (ClientSession session : sessions) - { - session.close(); - } - - for (int i = 0; i < numSessions; i++) - { - SimpleString subName = new SimpleString("sub" + i); - - s.deleteQueue(subName); - } - - s.close(); - - long end = System.currentTimeMillis(); - - log.info("duration " + (end - start)); - } - - // Now with synchronous receive() - - protected void doTestE(final ClientSessionFactory sf) throws Exception - { - long start = System.currentTimeMillis(); - - ClientSession s = sf.createSession(false, false, false); - - final int numMessages = 100; - - final int numSessions = 10; - - Set<ClientConsumer> consumers = new HashSet<ClientConsumer>(); - Set<ClientSession> sessions = new HashSet<ClientSession>(); - - for (int i = 0; i < numSessions; i++) - { - SimpleString subName = new SimpleString("sub" + i); - - ClientSession sessConsume = sf.createSession(false, true, true); - - sessConsume.start(); - - sessConsume.createQueue(ADDRESS, subName, null, false); - - ClientConsumer consumer = sessConsume.createConsumer(subName); - - consumers.add(consumer); - - sessions.add(sessConsume); - } - - ClientSession sessSend = sf.createSession(false, true, true); - - ClientProducer producer = sessSend.createProducer(ADDRESS); - - for (int i = 0; i < numMessages; i++) - { - ClientMessage message = sessSend.createClientMessage(HornetQTextMessage.TYPE, - false, - 0, - System.currentTimeMillis(), - (byte)1); - message.putIntProperty(new SimpleString("count"), i); - producer.send(message); - } - - for (int i = 0; i < numMessages; i++) - { - for (ClientConsumer consumer : consumers) - { - ClientMessage msg = consumer.receive(RECEIVE_TIMEOUT); - - assertNotNull(msg); - - assertEquals(i, msg.getProperty(new SimpleString("count"))); - - msg.acknowledge(); - } - } - - for (int i = 0; i < numMessages; i++) - { - for (ClientConsumer consumer : consumers) - { - ClientMessage msg = consumer.receiveImmediate(); - - assertNull(msg); - } - } - - sessSend.close(); - for (ClientSession session : sessions) - { - session.close(); - } - - for (int i = 0; i < numSessions; i++) - { - SimpleString subName = new SimpleString("sub" + i); - - s.deleteQueue(subName); - } - - s.close(); - - long end = System.currentTimeMillis(); - - log.info("duration " + (end - start)); - } - - protected void doTestF(final ClientSessionFactory sf) throws Exception - { - long start = System.currentTimeMillis(); - - ClientSession s = sf.createSession(false, false, false); - - final int numMessages = 100; - - final int numSessions = 10; - - Set<ClientConsumer> consumers = new HashSet<ClientConsumer>(); - Set<ClientSession> sessions = new HashSet<ClientSession>(); - - for (int i = 0; i < numSessions; i++) - { - SimpleString subName = new SimpleString("sub" + i); - - ClientSession sessConsume = sf.createSession(false, true, true); - - sessConsume.createQueue(ADDRESS, subName, null, false); - - ClientConsumer consumer = sessConsume.createConsumer(subName); - - consumers.add(consumer); - - sessions.add(sessConsume); - } - - ClientSession sessSend = sf.createSession(false, true, true); - - ClientProducer producer = sessSend.createProducer(ADDRESS); - - for (int i = 0; i < numMessages; i++) - { - ClientMessage message = sessSend.createClientMessage(HornetQTextMessage.TYPE, - false, - 0, - System.currentTimeMillis(), - (byte)1); - message.putIntProperty(new SimpleString("count"), i); - producer.send(message); - } - - for (ClientSession session : sessions) - { - session.start(); - } - - for (int i = 0; i < numMessages; i++) - { - for (ClientConsumer consumer : consumers) - { - ClientMessage msg = consumer.receive(RECEIVE_TIMEOUT); - - if (msg == null) - { - throw new IllegalStateException("Failed to receive message " + i); - } - - assertNotNull(msg); - - assertEquals(i, msg.getProperty(new SimpleString("count"))); - - msg.acknowledge(); - } - } - - for (int i = 0; i < numMessages; i++) - { - for (ClientConsumer consumer : consumers) - { - ClientMessage msg = consumer.receiveImmediate(); - - assertNull(msg); - } - } - - sessSend.close(); - for (ClientSession session : sessions) - { - session.close(); - } - - for (int i = 0; i < numSessions; i++) - { - SimpleString subName = new SimpleString("sub" + i); - - s.deleteQueue(subName); - } - - s.close(); - - assertEquals(1, ((ClientSessionFactoryImpl)sf).numSessions()); - - long end = System.currentTimeMillis(); - - log.info("duration " + (end - start)); - } - - protected void doTestG(final ClientSessionFactory sf) throws Exception - { - long start = System.currentTimeMillis(); - - ClientSession s = sf.createSession(false, false, false); - - final int numMessages = 100; - - final int numSessions = 10; - - Set<ClientConsumer> consumers = new HashSet<ClientConsumer>(); - Set<ClientSession> sessions = new HashSet<ClientSession>(); - - for (int i = 0; i < numSessions; i++) - { - SimpleString subName = new SimpleString("sub" + i); - - ClientSession sessConsume = sf.createSession(false, false, false); - - sessConsume.start(); - - sessConsume.createQueue(ADDRESS, subName, null, false); - - ClientConsumer consumer = sessConsume.createConsumer(subName); - - consumers.add(consumer); - - sessions.add(sessConsume); - } - - ClientSession sessSend = sf.createSession(false, false, false); - - ClientProducer producer = sessSend.createProducer(ADDRESS); - - for (int i = 0; i < numMessages; i++) - { - ClientMessage message = sessSend.createClientMessage(HornetQTextMessage.TYPE, - false, - 0, - System.currentTimeMillis(), - (byte)1); - message.putIntProperty(new SimpleString("count"), i); - producer.send(message); - } - - sessSend.rollback(); - - for (int i = 0; i < numMessages; i++) - { - ClientMessage message = sessSend.createClientMessage(HornetQTextMessage.TYPE, - false, - 0, - System.currentTimeMillis(), - (byte)1); - message.putIntProperty(new SimpleString("count"), i); - producer.send(message); - } - - sessSend.commit(); - - for (int i = 0; i < numMessages; i++) - { - for (ClientConsumer consumer : consumers) - { - ClientMessage msg = consumer.receive(RECEIVE_TIMEOUT); - - assertNotNull(msg); - - assertEquals(i, msg.getProperty(new SimpleString("count"))); - - msg.acknowledge(); - } - } - - for (ClientConsumer consumer : consumers) - { - ClientMessage msg = consumer.receiveImmediate(); - - assertNull(msg); - } - - for (ClientSession session : sessions) - { - session.rollback(); - } - - for (int i = 0; i < numMessages; i++) - { - for (ClientConsumer consumer : consumers) - { - ClientMessage msg = consumer.receive(RECEIVE_TIMEOUT); - - assertNotNull(msg); - - assertEquals(i, msg.getProperty(new SimpleString("count"))); - - msg.acknowledge(); - } - } - - for (int i = 0; i < numMessages; i++) - { - for (ClientConsumer consumer : consumers) - { - ClientMessage msg = consumer.receiveImmediate(); - - assertNull(msg); - } - } - - for (ClientSession session : sessions) - { - session.commit(); - } - - sessSend.close(); - for (ClientSession session : sessions) - { - session.close(); - } - - for (int i = 0; i < numSessions; i++) - { - SimpleString subName = new SimpleString("sub" + i); - - s.deleteQueue(subName); - } - - s.close(); - - long end = System.currentTimeMillis(); - - log.info("duration " + (end - start)); - } - - protected void doTestH(final ClientSessionFactory sf) throws Exception - { - long start = System.currentTimeMillis(); - - ClientSession s = sf.createSession(false, false, false); - - final int numMessages = 100; - - final int numSessions = 10; - - Set<ClientConsumer> consumers = new HashSet<ClientConsumer>(); - Set<ClientSession> sessions = new HashSet<ClientSession>(); - - for (int i = 0; i < numSessions; i++) - { - SimpleString subName = new SimpleString("sub" + i); - - ClientSession sessConsume = sf.createSession(false, false, false); - - sessConsume.createQueue(ADDRESS, subName, null, false); - - ClientConsumer consumer = sessConsume.createConsumer(subName); - - consumers.add(consumer); - - sessions.add(sessConsume); - } - - ClientSession sessSend = sf.createSession(false, false, false); - - ClientProducer producer = sessSend.createProducer(ADDRESS); - - for (int i = 0; i < numMessages; i++) - { - ClientMessage message = sessSend.createClientMessage(HornetQTextMessage.TYPE, - false, - 0, - System.currentTimeMillis(), - (byte)1); - message.putIntProperty(new SimpleString("count"), i); - producer.send(message); - } - - sessSend.rollback(); - - for (int i = 0; i < numMessages; i++) - { - ClientMessage message = sessSend.createClientMessage(HornetQTextMessage.TYPE, - false, - 0, - System.currentTimeMillis(), - (byte)1); - message.putIntProperty(new SimpleString("count"), i); - producer.send(message); - } - - sessSend.commit(); - - for (ClientSession session : sessions) - { - session.start(); - } - - for (int i = 0; i < numMessages; i++) - { - for (ClientConsumer consumer : consumers) - { - ClientMessage msg = consumer.receive(RECEIVE_TIMEOUT); - - assertNotNull(msg); - - assertEquals(i, msg.getProperty(new SimpleString("count"))); - - msg.acknowledge(); - } - } - - for (int i = 0; i < numMessages; i++) - { - for (ClientConsumer consumer : consumers) - { - ClientMessage msg = consumer.receiveImmediate(); - - assertNull(msg); - } - } - - for (ClientSession session : sessions) - { - session.rollback(); - } - - for (int i = 0; i < numMessages; i++) - { - for (ClientConsumer consumer : consumers) - { - ClientMessage msg = consumer.receive(RECEIVE_TIMEOUT); - - assertNotNull(msg); - - assertEquals(i, msg.getProperty(new SimpleString("count"))); - - msg.acknowledge(); - } - } - - for (int i = 0; i < numMessages; i++) - { - for (ClientConsumer consumer : consumers) - { - ClientMessage msg = consumer.receiveImmediate(); - - assertNull(msg); - } - } - - for (ClientSession session : sessions) - { - session.commit(); - } - - sessSend.close(); - for (ClientSession session : sessions) - { - session.close(); - } - - for (int i = 0; i < numSessions; i++) - { - SimpleString subName = new SimpleString("sub" + i); - - s.deleteQueue(subName); - } - - s.close(); - - long end = System.currentTimeMillis(); - - log.info("duration " + (end - start)); - } - - protected void doTestI(final ClientSessionFactory sf) throws Exception - { - ClientSession sessCreate = sf.createSession(false, true, true); - - sessCreate.createQueue(ADDRESS, ADDRESS, null, false); - - ClientSession sess = sf.createSession(false, true, true); - - sess.start(); - - ClientConsumer consumer = sess.createConsumer(ADDRESS); - - ClientProducer producer = sess.createProducer(ADDRESS); - - ClientMessage message = sess.createClientMessage(HornetQTextMessage.TYPE, - false, - 0, - System.currentTimeMillis(), - (byte)1); - producer.send(message); - - ClientMessage message2 = consumer.receive(RECEIVE_TIMEOUT); - - assertNotNull(message2); - - message2.acknowledge(); - - sess.close(); - - sessCreate.deleteQueue(ADDRESS); - - sessCreate.close(); - } - - protected void doTestJ(final ClientSessionFactory sf) throws Exception - { - ClientSession sessCreate = sf.createSession(false, true, true); - - sessCreate.createQueue(ADDRESS, ADDRESS, null, false); - - ClientSession sess = sf.createSession(false, true, true); - - sess.start(); - - ClientConsumer consumer = sess.createConsumer(ADDRESS); - - ClientProducer producer = sess.createProducer(ADDRESS); - - ClientMessage message = sess.createClientMessage(HornetQTextMessage.TYPE, - false, - 0, - System.currentTimeMillis(), - (byte)1); - producer.send(message); - - ClientMessage message2 = consumer.receive(RECEIVE_TIMEOUT); - - assertNotNull(message2); - - message2.acknowledge(); - - sess.close(); - - sessCreate.deleteQueue(ADDRESS); - - sessCreate.close(); - } - - protected void doTestK(final ClientSessionFactory sf) throws Exception - { - ClientSession s = sf.createSession(false, false, false); - - s.createQueue(ADDRESS, ADDRESS, null, false); - - final int numConsumers = 100; - - for (int i = 0; i < numConsumers; i++) - { - ClientConsumer consumer = s.createConsumer(ADDRESS); - - consumer.close(); - } - - s.deleteQueue(ADDRESS); - - s.close(); - } - - protected void doTestL(final ClientSessionFactory sf) throws Exception - { - final int numSessions = 10; - - for (int i = 0; i < numSessions; i++) - { - ClientSession session = sf.createSession(false, false, false); - - session.close(); - } - } - - protected void doTestN(final ClientSessionFactory sf) throws Exception - { - ClientSession sessCreate = sf.createSession(false, true, true); - - sessCreate.createQueue(ADDRESS, new SimpleString(ADDRESS.toString()), null, false); - - ClientSession sess = sf.createSession(false, true, true); - - sess.stop(); - - sess.start(); - - sess.stop(); - - ClientConsumer consumer = sess.createConsumer(new SimpleString(ADDRESS.toString())); - - ClientProducer producer = sess.createProducer(ADDRESS); - - ClientMessage message = sess.createClientMessage(HornetQTextMessage.TYPE, - false, - 0, - System.currentTimeMillis(), - (byte)1); - producer.send(message); - - sess.start(); - - ClientMessage message2 = consumer.receive(RECEIVE_TIMEOUT); - - assertNotNull(message2); - - message2.acknowledge(); - - sess.stop(); - - sess.start(); - - sess.close(); - - sessCreate.deleteQueue(new SimpleString(ADDRESS.toString())); - - sessCreate.close(); - } - - protected int getNumIterations() - { - return 2; - } - - protected void setUp() throws Exception - { - super.setUp(); - - timer = new Timer(true); - } - - protected void tearDown() throws Exception - { - timer.cancel(); - - InVMRegistry.instance.clear(); - - super.tearDown(); - } - - // Private ------------------------------------------------------- - - private Failer startFailer(final long time, final ClientSession session) - { - Failer failer = new Failer((ClientSessionInternal)session); - - timer.schedule(failer, (long)(time * Math.random()), 100); - - return failer; - } - - private void start() throws Exception - { - Configuration liveConf = new ConfigurationImpl(); - liveConf.setSecurityEnabled(false); - liveConf.getAcceptorConfigurations() - .add(new TransportConfiguration("org.hornetq.core.remoting.impl.invm.InVMAcceptorFactory")); - Map<String, TransportConfiguration> connectors = new HashMap<String, TransportConfiguration>(); - TransportConfiguration backupTC = new TransportConfiguration("org.hornetq.core.remoting.impl.invm.InVMConnectorFactory"); - connectors.put(backupTC.getName(), backupTC); - liveConf.setConnectorConfigurations(connectors); - liveConf.setBackupConnectorName(backupTC.getName()); - liveService = HornetQ.newHornetQServer(liveConf, false); - liveService.start(); - } - - private void stop() throws Exception - { - liveService.stop(); - - assertEquals(0, InVMRegistry.instance.size()); - - liveService = null; - } - - // Inner classes ------------------------------------------------- - - class Failer extends TimerTask - { - private final ClientSessionInternal session; - - private boolean executed; - - public Failer(final ClientSessionInternal session) - { - this.session = session; - } - - public synchronized void run() - { - log.info("** Failing connection"); - - session.getConnection().fail(new HornetQException(HornetQException.NOT_CONNECTED, "oops")); - - log.info("** Fail complete"); - - cancel(); - - executed = true; - } - - public synchronized boolean isExecuted() - { - return executed; - } - } - - public abstract class RunnableT - { - abstract void run(final ClientSessionFactory sf) throws Exception; - } -} Deleted: branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/integration/cluster/failover/ReattachTest.java =================================================================== --- branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/integration/cluster/failover/ReattachTest.java 2009-09-22 13:57:55 UTC (rev 7978) +++ branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/integration/cluster/failover/ReattachTest.java 2009-09-22 15:08:41 UTC (rev 7979) @@ -1,715 +0,0 @@ -/* - * Copyright 2009 Red Hat, Inc. - * Red Hat licenses this file to you under the Apache License, version - * 2.0 (the "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * http://www.apache.org/licenses/LICENSE-2.0 - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or - * implied. See the License for the specific language governing - * permissions and limitations under the License. - */ - -package org.hornetq.tests.integration.cluster.failover; - -import org.hornetq.core.client.ClientConsumer; -import org.hornetq.core.client.ClientMessage; -import org.hornetq.core.client.ClientProducer; -import org.hornetq.core.client.ClientSession; -import org.hornetq.core.client.impl.ClientSessionFactoryImpl; -import org.hornetq.core.client.impl.ClientSessionFactoryInternal; -import org.hornetq.core.client.impl.ClientSessionInternal; -import org.hornetq.core.config.Configuration; -import org.hornetq.core.config.TransportConfiguration; -import org.hornetq.core.config.impl.ConfigurationImpl; -import org.hornetq.core.exception.HornetQException; -import org.hornetq.core.logging.Logger; -import org.hornetq.core.remoting.FailureListener; -import org.hornetq.core.remoting.RemotingConnection; -import org.hornetq.core.remoting.impl.invm.InVMConnector; -import org.hornetq.core.remoting.impl.invm.InVMRegistry; -import org.hornetq.core.server.HornetQ; -import org.hornetq.core.server.HornetQServer; -import org.hornetq.jms.client.HornetQTextMessage; -import org.hornetq.tests.util.UnitTestCase; -import org.hornetq.utils.SimpleString; - -/** - * - * A ReattachTest - * - * @author <a href="mailto:tim.fox@jboss.com">Tim Fox</a> - * - * Created 4 Nov 2008 16:54:50 - * - * - */ -public class ReattachTest extends UnitTestCase -{ - private static final Logger log = Logger.getLogger(ReattachTest.class); - - // Constants ----------------------------------------------------- - - // Attributes ---------------------------------------------------- - - private static final SimpleString ADDRESS = new SimpleString("FailoverTestAddress"); - - private HornetQServer service; - - // Static -------------------------------------------------------- - - // Constructors -------------------------------------------------- - - // Public -------------------------------------------------------- - - /* - * Test failure on connection, but server is still up so should immediately reconnect - */ - public void testImmediateReattach() throws Exception - { - final long retryInterval = 500; - - final double retryMultiplier = 1d; - - final int reconnectAttempts = 1; - - ClientSessionFactoryInternal sf = new ClientSessionFactoryImpl(new TransportConfiguration("org.hornetq.core.remoting.impl.invm.InVMConnectorFactory")); - - sf.setRetryInterval(retryInterval); - sf.setRetryIntervalMultiplier(retryMultiplier); - sf.setReconnectAttempts(reconnectAttempts); - - ClientSession session = sf.createSession(false, true, true); - - session.createQueue(ADDRESS, ADDRESS, null, false); - - final int numIterations = 100; - - for (int j = 0; j < numIterations; j++) - { - ClientProducer producer = session.createProducer(ADDRESS); - - final int numMessages = 1000; - - for (int i = 0; i < numMessages; i++) - { - ClientMessage message = session.createClientMessage(HornetQTextMessage.TYPE, - false, - 0, - System.currentTimeMillis(), - (byte)1); - message.putIntProperty(new SimpleString("count"), i); - message.getBody().writeString("aardvarks"); - producer.send(message); - } - - ClientConsumer consumer = session.createConsumer(ADDRESS); - - RemotingConnection conn = ((ClientSessionInternal)session).getConnection(); - - conn.fail(new HornetQException(HornetQException.NOT_CONNECTED)); - - session.start(); - - for (int i = 0; i < numMessages; i++) - { - ClientMessage message = consumer.receive(500); - - assertNotNull(message); - - assertEquals("aardvarks", message.getBody().readString()); - - assertEquals(i, message.getProperty(new SimpleString("count"))); - - message.acknowledge(); - } - - ClientMessage message = consumer.receiveImmediate(); - - assertNull(message); - - producer.close(); - - consumer.close(); - } - - session.close(); - - sf.close(); - } - - /* - * Test failure on connection, simulate failure to create connection for a while, then - * allow connection to be recreated - */ - public void testDelayedReattach() throws Exception - { - final long retryInterval = 500; - - final double retryMultiplier = 1d; - - final int reconnectAttempts = -1; - - ClientSessionFactoryInternal sf = new ClientSessionFactoryImpl(new TransportConfiguration("org.hornetq.core.remoting.impl.invm.InVMConnectorFactory")); - - sf.setRetryInterval(retryInterval); - sf.setRetryIntervalMultiplier(retryMultiplier); - sf.setReconnectAttempts(reconnectAttempts); - - ClientSession session = sf.createSession(false, true, true); - - session.createQueue(ADDRESS, ADDRESS, null, false); - - ClientProducer producer = session.createProducer(ADDRESS); - - final int numMessages = 1000; - - for (int i = 0; i < numMessages; i++) - { - ClientMessage message = session.createClientMessage(HornetQTextMessage.TYPE, - false, - 0, - System.currentTimeMillis(), - (byte)1); - message.putIntProperty(new SimpleString("count"), i); - message.getBody().writeString("aardvarks"); - producer.send(message); - } - - ClientConsumer consumer = session.createConsumer(ADDRESS); - - InVMConnector.failOnCreateConnection = true; - - RemotingConnection conn = ((ClientSessionInternal)session).getConnection(); - - Thread t = new Thread() - { - public void run() - { - try - { - Thread.sleep(retryInterval * 3); - } - catch (InterruptedException ignore) - { - } - - InVMConnector.failOnCreateConnection = false; - } - }; - - t.start(); - - conn.fail(new HornetQException(HornetQException.NOT_CONNECTED)); - - session.start(); - - for (int i = 0; i < numMessages; i++) - { - ClientMessage message = consumer.receive(500); - - assertNotNull(message); - - assertEquals("aardvarks", message.getBody().readString()); - - assertEquals(i, message.getProperty(new SimpleString("count"))); - - message.acknowledge(); - } - - ClientMessage message = consumer.receiveImmediate(); - - assertNull(message); - - session.close(); - - sf.close(); - - t.join(); - } - - // Test an async (e.g. pinger) failure coming in while a connection manager is already reconnecting - public void testAsyncFailureWhileReattaching() throws Exception - { - final long retryInterval = 500; - - final double retryMultiplier = 1d; - - final int reconnectAttempts = -1; - - final long asyncFailDelay = 2000; - - ClientSessionFactoryInternal sf = new ClientSessionFactoryImpl(new TransportConfiguration("org.hornetq.core.remoting.impl.invm.InVMConnectorFactory")); - - sf.setRetryInterval(retryInterval); - sf.setRetryIntervalMultiplier(retryMultiplier); - sf.setReconnectAttempts(reconnectAttempts); - - ClientSession session = sf.createSession(false, true, true); - - ClientSession session2 = sf.createSession(false, true, true); - - class MyFailureListener implements FailureListener - { - volatile boolean failed; - - public void connectionFailed(HornetQException me) - { - failed = true; - } - } - - MyFailureListener listener = new MyFailureListener(); - - session2.addFailureListener(listener); - - session.createQueue(ADDRESS, ADDRESS, null, false); - - ClientProducer producer = session.createProducer(ADDRESS); - - final int numMessages = 1000; - - for (int i = 0; i < numMessages; i++) - { - ClientMessage message = session.createClientMessage(HornetQTextMessage.TYPE, - false, - 0, - System.currentTimeMillis(), - (byte)1); - message.putIntProperty(new SimpleString("count"), i); - message.getBody().writeString("aardvarks"); - producer.send(message); - } - - ClientConsumer consumer = session.createConsumer(ADDRESS); - - InVMConnector.numberOfFailures = 10; - InVMConnector.failOnCreateConnection = true; - - //We need to fail on different connections. - - //We fail on one connection then the connection manager tries to reconnect all connections - //Then we fail the other, and the connection manager is then called while the reconnection is occurring - //We can't use the same connection since RemotingConnectionImpl only allows one fail to be in process - //at same time - - final RemotingConnection conn = ((ClientSessionInternal)session).getConnection(); - - final RemotingConnection conn2 = ((ClientSessionInternal)session2).getConnection(); - - assertTrue(conn != conn2); - - Thread t = new Thread() - { - public void run() - { - try - { - Thread.sleep(asyncFailDelay); - } - catch (InterruptedException ignore) - { - } - - log.info("calling fail async"); - - conn2.fail(new HornetQException(HornetQException.NOT_CONNECTED, "Did not receive pong from server")); - } - }; - - t.start(); - - conn.fail(new HornetQException(HornetQException.NOT_CONNECTED)); - - assertFalse(listener.failed); - - session.start(); - - for (int i = 0; i < numMessages; i++) - { - ClientMessage message = consumer.receive(500); - - assertNotNull(message); - - assertEquals("aardvarks", message.getBody().readString()); - - assertEquals(i, message.getProperty(new SimpleString("count"))); - - message.acknowledge(); - } - - ClientMessage message = consumer.receiveImmediate(); - - assertNull(message); - - session.close(); - - session2.close(); - - sf.close(); - - t.join(); - } - - public void testReattachAttemptsFailsToReconnect() throws Exception - { - final long retryInterval = 500; - - final double retryMultiplier = 1d; - - final int reconnectAttempts = 3; - - ClientSessionFactoryInternal sf = new ClientSessionFactoryImpl(new TransportConfiguration("org.hornetq.core.remoting.impl.invm.InVMConnectorFactory")); - - sf.setRetryInterval(retryInterval); - sf.setRetryIntervalMultiplier(retryMultiplier); - sf.setReconnectAttempts(reconnectAttempts); - - ClientSession session = sf.createSession(false, true, true); - - session.createQueue(ADDRESS, ADDRESS, null, false); - - ClientProducer producer = session.createProducer(ADDRESS); - - final int numMessages = 1000; - - for (int i = 0; i < numMessages; i++) - { - ClientMessage message = session.createClientMessage(HornetQTextMessage.TYPE, - false, - 0, - System.currentTimeMillis(), - (byte)1); - message.putIntProperty(new SimpleString("count"), i); - message.getBody().writeString("aardvarks"); - producer.send(message); - } - - ClientConsumer consumer = session.createConsumer(ADDRESS); - - InVMConnector.failOnCreateConnection = true; - - RemotingConnection conn = ((ClientSessionInternal)session).getConnection(); - - // Sleep for longer than max retries so should fail to reconnect - - Thread t = new Thread() - { - public void run() - { - try - { - Thread.sleep(retryInterval * (reconnectAttempts + 1)); - } - catch (InterruptedException ignore) - { - } - - InVMConnector.failOnCreateConnection = false; - } - }; - - t.start(); - - conn.fail(new HornetQException(HornetQException.NOT_CONNECTED)); - - session.start(); - - // Should be null since failed to reconnect - ClientMessage message = consumer.receive(500); - - assertNull(message); - - session.close(); - - sf.close(); - - t.join(); - } - - public void testReattachAttemptsSucceedsInReconnecting() throws Exception - { - final long retryInterval = 500; - - final double retryMultiplier = 1d; - - final int reconnectAttempts = 10; - - ClientSessionFactoryInternal sf = new ClientSessionFactoryImpl(new TransportConfiguration("org.hornetq.core.remoting.impl.invm.InVMConnectorFactory")); - - sf.setRetryInterval(retryInterval); - sf.setRetryIntervalMultiplier(retryMultiplier); - sf.setReconnectAttempts(reconnectAttempts); - - ClientSession session = sf.createSession(false, true, true); - - session.createQueue(ADDRESS, ADDRESS, null, false); - - ClientProducer producer = session.createProducer(ADDRESS); - - final int numMessages = 1000; - - for (int i = 0; i < numMessages; i++) - { - ClientMessage message = session.createClientMessage(HornetQTextMessage.TYPE, - false, - 0, - System.currentTimeMillis(), - (byte)1); - message.putIntProperty(new SimpleString("count"), i); - message.getBody().writeString("aardvarks"); - producer.send(message); - } - - ClientConsumer consumer = session.createConsumer(ADDRESS); - - InVMConnector.failOnCreateConnection = true; - InVMConnector.numberOfFailures = reconnectAttempts - 1; - - RemotingConnection conn = ((ClientSessionInternal)session).getConnection(); - - conn.fail(new HornetQException(HornetQException.NOT_CONNECTED)); - - session.start(); - - for (int i = 0; i < numMessages; i++) - { - ClientMessage message = consumer.receive(500); - - assertNotNull(message); - - assertEquals("aardvarks", message.getBody().readString()); - - assertEquals(i, message.getProperty(new SimpleString("count"))); - - message.acknowledge(); - } - - ClientMessage message = consumer.receiveImmediate(); - - assertNull(message); - - session.close(); - - sf.close(); - } - - public void testRetryInterval() throws Exception - { - final long retryInterval = 500; - - final double retryMultiplier = 1d; - - final int reconnectAttempts = -1; - - ClientSessionFactoryInternal sf = new ClientSessionFactoryImpl(new TransportConfiguration("org.hornetq.core.remoting.impl.invm.InVMConnectorFactory")); - - sf.setRetryInterval(retryInterval); - sf.setRetryIntervalMultiplier(retryMultiplier); - sf.setReconnectAttempts(reconnectAttempts); - - ClientSession session = sf.createSession(false, true, true); - - session.createQueue(ADDRESS, ADDRESS, null, false); - - ClientProducer producer = session.createProducer(ADDRESS); - - final int numMessages = 1000; - - for (int i = 0; i < numMessages; i++) - { - ClientMessage message = session.createClientMessage(HornetQTextMessage.TYPE, - false, - 0, - System.currentTimeMillis(), - (byte)1); - message.putIntProperty(new SimpleString("count"), i); - message.getBody().writeString("aardvarks"); - producer.send(message); - } - - ClientConsumer consumer = session.createConsumer(ADDRESS); - - InVMConnector.failOnCreateConnection = true; - - RemotingConnection conn = ((ClientSessionInternal)session).getConnection(); - - long start = System.currentTimeMillis(); - - Thread t = new Thread() - { - public void run() - { - try - { - Thread.sleep(retryInterval / 2); - } - catch (InterruptedException ignore) - { - } - InVMConnector.failOnCreateConnection = false; - } - }; - - t.start(); - - conn.fail(new HornetQException(HornetQException.NOT_CONNECTED)); - - session.start(); - - for (int i = 0; i < numMessages; i++) - { - ClientMessage message = consumer.receive(500); - - assertNotNull(message); - - assertEquals("aardvarks", message.getBody().readString()); - - assertEquals(i, message.getProperty(new SimpleString("count"))); - - message.acknowledge(); - } - - ClientMessage message = consumer.receiveImmediate(); - - assertNull(message); - - long end = System.currentTimeMillis(); - - assertTrue((end - start) >= retryInterval); - - session.close(); - - sf.close(); - - t.join(); - } - - public void testExponentialBackoff() throws Exception - { - final long retryInterval = 500; - - final double retryMultiplier = 4d; - - final int reconnectAttempts = -1; - - ClientSessionFactoryInternal sf = new ClientSessionFactoryImpl(new TransportConfiguration("org.hornetq.core.remoting.impl.invm.InVMConnectorFactory")); - - sf.setRetryInterval(retryInterval); - sf.setRetryIntervalMultiplier(retryMultiplier); - sf.setReconnectAttempts(reconnectAttempts); - - ClientSession session = sf.createSession(false, true, true); - - session.createQueue(ADDRESS, ADDRESS, null, false); - - ClientProducer producer = session.createProducer(ADDRESS); - - final int numMessages = 1000; - - for (int i = 0; i < numMessages; i++) - { - ClientMessage message = session.createClientMessage(HornetQTextMessage.TYPE, - false, - 0, - System.currentTimeMillis(), - (byte)1); - message.putIntProperty(new SimpleString("count"), i); - message.getBody().writeString("aardvarks"); - producer.send(message); - } - - ClientConsumer consumer = session.createConsumer(ADDRESS); - - InVMConnector.failOnCreateConnection = true; - - RemotingConnection conn = ((ClientSessionInternal)session).getConnection(); - - long start = System.currentTimeMillis(); - - Thread t = new Thread() - { - public void run() - { - try - { - Thread.sleep(retryInterval * 2); - } - catch (InterruptedException ignore) - { - } - - InVMConnector.failOnCreateConnection = false; - } - }; - - t.start(); - - conn.fail(new HornetQException(HornetQException.NOT_CONNECTED)); - - session.start(); - - for (int i = 0; i < numMessages; i++) - { - ClientMessage message = consumer.receive(500); - - assertNotNull(message); - - assertEquals("aardvarks", message.getBody().readString()); - - assertEquals(i, message.getProperty(new SimpleString("count"))); - - message.acknowledge(); - } - - ClientMessage message = consumer.receiveImmediate(); - - assertNull(message); - - long end = System.currentTimeMillis(); - - assertTrue((end - start) >= retryInterval * (1 + retryMultiplier)); - - session.close(); - - sf.close(); - - t.join(); - } - - // Package protected --------------------------------------------- - - // Protected ----------------------------------------------------- - - @Override - protected void setUp() throws Exception - { - super.setUp(); - - Configuration liveConf = new ConfigurationImpl(); - liveConf.setSecurityEnabled(false); - liveConf.getAcceptorConfigurations() - .add(new TransportConfiguration("org.hornetq.core.remoting.impl.invm.InVMAcceptorFactory")); - service = HornetQ.newHornetQServer(liveConf, false); - service.start(); - } - - @Override - protected void tearDown() throws Exception - { - InVMConnector.resetFailures(); - - service.stop(); - - assertEquals(0, InVMRegistry.instance.size()); - - service = null; - - super.tearDown(); - } - - // Private ------------------------------------------------------- - - // Inner classes ------------------------------------------------- -} Copied: branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/integration/cluster/reattach/MultiThreadRandomReattachTest.java (from rev 7962, branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/integration/cluster/failover/MultiThreadRandomReattachTest.java) =================================================================== --- branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/integration/cluster/reattach/MultiThreadRandomReattachTest.java (rev 0) +++ branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/integration/cluster/reattach/MultiThreadRandomReattachTest.java 2009-09-22 15:08:41 UTC (rev 7979) @@ -0,0 +1,61 @@ +/* + * Copyright 2009 Red Hat, Inc. + * Red Hat licenses this file to you under the Apache License, version + * 2.0 (the "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * http://www.apache.org/licenses/LICENSE-2.0 + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or + * implied. See the License for the specific language governing + * permissions and limitations under the License. + */ + +package org.hornetq.tests.integration.cluster.reattach; + +import org.hornetq.core.client.ClientMessage; +import org.hornetq.core.config.Configuration; +import org.hornetq.core.config.TransportConfiguration; +import org.hornetq.core.config.impl.ConfigurationImpl; +import org.hornetq.core.server.HornetQ; + + +/** + * + * A MultiThreadRandomReattachTest + * + * @author <a href="mailto:tim.fox@jboss.com">Tim Fox</a> + * + * + */ +public class MultiThreadRandomReattachTest extends MultiThreadRandomReattachTestBase +{ + @Override + protected void start() throws Exception + { + Configuration liveConf = new ConfigurationImpl(); + liveConf.setSecurityEnabled(false); + liveConf.getAcceptorConfigurations() + .add(new TransportConfiguration("org.hornetq.core.remoting.impl.invm.InVMAcceptorFactory")); + liveServer = HornetQ.newHornetQServer(liveConf, false); + liveServer.start(); + } + + /* (non-Javadoc) + * @see org.hornetq.tests.integration.cluster.failover.MultiThreadRandomReattachTestBase#setBody(org.hornetq.core.client.ClientMessage) + */ + @Override + protected void setBody(final ClientMessage message) throws Exception + { + } + + /* (non-Javadoc) + * @see org.hornetq.tests.integration.cluster.failover.MultiThreadRandomReattachTestBase#checkSize(org.hornetq.core.client.ClientMessage) + */ + @Override + protected boolean checkSize(final ClientMessage message) + { + return 0 == message.getBody().writerIndex(); + } + +} Copied: branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/integration/cluster/reattach/MultiThreadRandomReattachTestBase.java (from rev 7962, branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/integration/cluster/failover/MultiThreadRandomReattachTestBase.java) =================================================================== --- branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/integration/cluster/reattach/MultiThreadRandomReattachTestBase.java (rev 0) +++ branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/integration/cluster/reattach/MultiThreadRandomReattachTestBase.java 2009-09-22 15:08:41 UTC (rev 7979) @@ -0,0 +1,1500 @@ +/* + * Copyright 2009 Red Hat, Inc. + * Red Hat licenses this file to you under the Apache License, version + * 2.0 (the "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * http://www.apache.org/licenses/LICENSE-2.0 + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or + * implied. See the License for the specific language governing + * permissions and limitations under the License. + */ + +package org.hornetq.tests.integration.cluster.reattach; + +import java.util.HashMap; +import java.util.HashSet; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; + +import org.hornetq.core.client.ClientConsumer; +import org.hornetq.core.client.ClientMessage; +import org.hornetq.core.client.ClientProducer; +import org.hornetq.core.client.ClientSession; +import org.hornetq.core.client.ClientSessionFactory; +import org.hornetq.core.client.MessageHandler; +import org.hornetq.core.client.impl.ClientSessionFactoryImpl; +import org.hornetq.core.client.impl.ClientSessionFactoryInternal; +import org.hornetq.core.config.TransportConfiguration; +import org.hornetq.core.exception.HornetQException; +import org.hornetq.core.logging.Logger; +import org.hornetq.core.remoting.impl.invm.InVMRegistry; +import org.hornetq.core.server.HornetQServer; +import org.hornetq.jms.client.HornetQBytesMessage; +import org.hornetq.jms.client.HornetQTextMessage; +import org.hornetq.utils.SimpleString; + +/** + * A MultiThreadRandomReattachTestBase + * + * @author <a href="mailto:tim.fox@jboss.com">Tim Fox</a> + * @author <a href="mailto:clebert.suconic@jboss.org">Clebert Suconic</a> + * + * + */ +public abstract class MultiThreadRandomReattachTestBase extends MultiThreadReattachSupport +{ + private final Logger log = Logger.getLogger(getClass()); + + // Constants ----------------------------------------------------- + + private static final int RECEIVE_TIMEOUT = 30000; + + private final int LATCH_WAIT = getLatchWait(); + + private int NUM_THREADS = getNumThreads(); + + // Attributes ---------------------------------------------------- + protected static final SimpleString ADDRESS = new SimpleString("FailoverTestAddress"); + + protected HornetQServer liveServer; + + // Static -------------------------------------------------------- + + // Constructors -------------------------------------------------- + + // Public -------------------------------------------------------- + + public void testA() throws Exception + { + runTestMultipleThreads(new RunnableT() + { + @Override + public void run(final ClientSessionFactory sf, final int threadNum) throws Exception + { + doTestA(sf, threadNum); + } + }, NUM_THREADS, false); + + } + + public void testB() throws Exception + { + runTestMultipleThreads(new RunnableT() + { + @Override + public void run(final ClientSessionFactory sf, final int threadNum) throws Exception + { + doTestB(sf, threadNum); + } + }, NUM_THREADS, false); + } + + public void testC() throws Exception + { + runTestMultipleThreads(new RunnableT() + { + @Override + public void run(final ClientSessionFactory sf, final int threadNum) throws Exception + { + doTestC(sf, threadNum); + } + }, NUM_THREADS, false); + } + + public void testD() throws Exception + { + runTestMultipleThreads(new RunnableT() + { + @Override + public void run(final ClientSessionFactory sf, final int threadNum) throws Exception + { + doTestD(sf, threadNum); + } + }, NUM_THREADS, false); + } + + public void testE() throws Exception + { + runTestMultipleThreads(new RunnableT() + { + @Override + public void run(final ClientSessionFactory sf, final int threadNum) throws Exception + { + doTestE(sf, threadNum); + } + }, NUM_THREADS, false); + } + + public void testF() throws Exception + { + runTestMultipleThreads(new RunnableT() + { + @Override + public void run(final ClientSessionFactory sf, final int threadNum) throws Exception + { + doTestF(sf, threadNum); + } + }, NUM_THREADS, false); + } + + public void testG() throws Exception + { + runTestMultipleThreads(new RunnableT() + { + @Override + public void run(final ClientSessionFactory sf, final int threadNum) throws Exception + { + doTestG(sf, threadNum); + } + }, NUM_THREADS, false); + } + + public void testH() throws Exception + { + runTestMultipleThreads(new RunnableT() + { + @Override + public void run(final ClientSessionFactory sf, final int threadNum) throws Exception + { + doTestH(sf, threadNum); + } + }, NUM_THREADS, false); + } + + public void testI() throws Exception + { + runTestMultipleThreads(new RunnableT() + { + @Override + public void run(final ClientSessionFactory sf, final int threadNum) throws Exception + { + doTestI(sf, threadNum); + } + }, NUM_THREADS, false); + } + + public void testJ() throws Exception + { + runTestMultipleThreads(new RunnableT() + { + @Override + public void run(final ClientSessionFactory sf, final int threadNum) throws Exception + { + doTestJ(sf, threadNum); + } + }, NUM_THREADS, false); + } + + public void testK() throws Exception + { + runTestMultipleThreads(new RunnableT() + { + @Override + public void run(final ClientSessionFactory sf, final int threadNum) throws Exception + { + doTestK(sf, threadNum); + } + }, NUM_THREADS, false); + } + + public void testL() throws Exception + { + runTestMultipleThreads(new RunnableT() + { + @Override + public void run(final ClientSessionFactory sf, final int threadNum) throws Exception + { + doTestL(sf); + } + }, NUM_THREADS, true, 10); + } + + // public void testM() throws Exception + // { + // runTestMultipleThreads(new RunnableT() + // { + // public void run(final ClientSessionFactory sf, final int threadNum) throws Exception + // { + // doTestM(sf, threadNum); + // } + // }, NUM_THREADS); + // } + + public void testN() throws Exception + { + runTestMultipleThreads(new RunnableT() + { + @Override + public void run(final ClientSessionFactory sf, final int threadNum) throws Exception + { + doTestN(sf, threadNum); + } + }, NUM_THREADS, false); + } + + // Package protected --------------------------------------------- + + // Protected ----------------------------------------------------- + + protected abstract void start() throws Exception; + + protected abstract void setBody(ClientMessage message) throws Exception; + + protected abstract boolean checkSize(ClientMessage message); + + protected int getNumThreads() + { + return 10; + } + + protected ClientSession createAutoCommitSession(ClientSessionFactory sf) throws Exception + { + return sf.createSession(false, true, true); + } + + protected ClientSession createTransactionalSession(ClientSessionFactory sf) throws Exception + { + return sf.createSession(false, false, false); + } + + protected void doTestA(final ClientSessionFactory sf, final int threadNum, final ClientSession session2) throws Exception + { + SimpleString subName = new SimpleString("sub" + threadNum); + + ClientSession session = sf.createSession(false, true, true); + + session.createQueue(ADDRESS, subName, null, false); + + ClientProducer producer = session.createProducer(ADDRESS); + + ClientConsumer consumer = session.createConsumer(subName); + + final int numMessages = 100; + + sendMessages(session, producer, numMessages, threadNum); + + session.start(); + + MyHandler handler = new MyHandler(threadNum, numMessages); + + consumer.setMessageHandler(handler); + + boolean ok = handler.latch.await(LATCH_WAIT, TimeUnit.MILLISECONDS); + + if (!ok) + { + throw new Exception("Timed out waiting for messages on handler " + System.identityHashCode(handler) + + " threadnum " + + threadNum); + } + + if (handler.failure != null) + { + throw new Exception("Handler failed: " + handler.failure); + } + + producer.close(); + + consumer.close(); + + session.deleteQueue(subName); + + session.close(); + } + + protected void doTestA(final ClientSessionFactory sf, final int threadNum) throws Exception + { + long start = System.currentTimeMillis(); + + ClientSession s = sf.createSession(false, false, false); + + final int numMessages = 100; + + final int numSessions = 10; + + Set<ClientConsumer> consumers = new HashSet<ClientConsumer>(); + Set<ClientSession> sessions = new HashSet<ClientSession>(); + + for (int i = 0; i < numSessions; i++) + { + SimpleString subName = new SimpleString(threadNum + "sub" + i); + + ClientSession sessConsume = createAutoCommitSession(sf); + + sessConsume.start(); + + sessConsume.createQueue(ADDRESS, subName, null, false); + + ClientConsumer consumer = sessConsume.createConsumer(subName); + + consumers.add(consumer); + + sessions.add(sessConsume); + + } + + ClientSession sessSend = sf.createSession(false, true, true); + + ClientProducer producer = sessSend.createProducer(ADDRESS); + + sendMessages(sessSend, producer, numMessages, threadNum); + + Set<MyHandler> handlers = new HashSet<MyHandler>(); + + for (ClientConsumer consumer : consumers) + { + MyHandler handler = new MyHandler(threadNum, numMessages); + + consumer.setMessageHandler(handler); + + handlers.add(handler); + } + + for (MyHandler handler : handlers) + { + boolean ok = handler.latch.await(LATCH_WAIT, TimeUnit.MILLISECONDS); + + if (!ok) + { + throw new Exception("Timed out waiting for messages on handler " + System.identityHashCode(handler) + + " threadnum " + + threadNum); + } + + if (handler.failure != null) + { + throw new Exception("Handler failed: " + handler.failure); + } + } + + sessSend.close(); + + for (ClientSession session : sessions) + { + session.close(); + } + + for (int i = 0; i < numSessions; i++) + { + SimpleString subName = new SimpleString(threadNum + "sub" + i); + + s.deleteQueue(subName); + } + + s.close(); + + long end = System.currentTimeMillis(); + + log.info("duration " + (end - start)); + } + + protected void doTestB(final ClientSessionFactory sf, final int threadNum) throws Exception + { + long start = System.currentTimeMillis(); + + ClientSession s = sf.createSession(false, false, false); + + final int numMessages = 100; + + final int numSessions = 10; + + Set<ClientConsumer> consumers = new HashSet<ClientConsumer>(); + Set<ClientSession> sessions = new HashSet<ClientSession>(); + + for (int i = 0; i < numSessions; i++) + { + SimpleString subName = new SimpleString(threadNum + "sub" + i); + + ClientSession sessConsume = createAutoCommitSession(sf); + + sessConsume.createQueue(ADDRESS, subName, null, false); + + ClientConsumer consumer = sessConsume.createConsumer(subName); + + consumers.add(consumer); + + sessions.add(sessConsume); + } + + ClientSession sessSend = sf.createSession(false, true, true); + + ClientProducer producer = sessSend.createProducer(ADDRESS); + + sendMessages(sessSend, producer, numMessages, threadNum); + + for (ClientSession session : sessions) + { + session.start(); + } + + Set<MyHandler> handlers = new HashSet<MyHandler>(); + + for (ClientConsumer consumer : consumers) + { + MyHandler handler = new MyHandler(threadNum, numMessages); + + consumer.setMessageHandler(handler); + + handlers.add(handler); + } + + for (MyHandler handler : handlers) + { + boolean ok = handler.latch.await(LATCH_WAIT, TimeUnit.MILLISECONDS); + + if (!ok) + { + throw new Exception("Timed out waiting for messages on handler " + System.identityHashCode(handler) + + " threadnum " + + threadNum); + } + + if (handler.failure != null) + { + throw new Exception("Handler failed: " + handler.failure); + } + } + + sessSend.close(); + + for (ClientSession session : sessions) + { + session.close(); + } + + for (int i = 0; i < numSessions; i++) + { + SimpleString subName = new SimpleString(threadNum + "sub" + i); + + s.deleteQueue(subName); + } + + s.close(); + + long end = System.currentTimeMillis(); + + log.info("duration " + (end - start)); + + } + + protected void doTestC(final ClientSessionFactory sf, final int threadNum) throws Exception + { + long start = System.currentTimeMillis(); + + ClientSession s = sf.createSession(false, false, false); + + final int numMessages = 100; + + final int numSessions = 10; + + Set<ClientConsumer> consumers = new HashSet<ClientConsumer>(); + Set<ClientSession> sessions = new HashSet<ClientSession>(); + + for (int i = 0; i < numSessions; i++) + { + SimpleString subName = new SimpleString(threadNum + "sub" + i); + + ClientSession sessConsume = createTransactionalSession(sf); + + sessConsume.start(); + + sessConsume.createQueue(ADDRESS, subName, null, false); + + ClientConsumer consumer = sessConsume.createConsumer(subName); + + consumers.add(consumer); + + sessions.add(sessConsume); + } + + ClientSession sessSend = sf.createSession(false, false, false); + + ClientProducer producer = sessSend.createProducer(ADDRESS); + + sendMessages(sessSend, producer, numMessages, threadNum); + + sessSend.rollback(); + + sendMessages(sessSend, producer, numMessages, threadNum); + + sessSend.commit(); + + Set<MyHandler> handlers = new HashSet<MyHandler>(); + + for (ClientConsumer consumer : consumers) + { + MyHandler handler = new MyHandler(threadNum, numMessages); + + consumer.setMessageHandler(handler); + + handlers.add(handler); + } + + for (MyHandler handler : handlers) + { + boolean ok = handler.latch.await(LATCH_WAIT, TimeUnit.MILLISECONDS); + + if (!ok) + { + throw new Exception("Timed out waiting for messages on handler " + System.identityHashCode(handler) + + " threadnum " + + threadNum); + } + + if (handler.failure != null) + { + throw new Exception("Handler failed: " + handler.failure); + } + + handler.reset(); + } + + for (ClientSession session : sessions) + { + session.rollback(); + } + + for (MyHandler handler : handlers) + { + boolean ok = handler.latch.await(LATCH_WAIT, TimeUnit.MILLISECONDS); + + assertTrue(ok); + } + + for (ClientSession session : sessions) + { + session.commit(); + } + + sessSend.close(); + for (ClientSession session : sessions) + { + session.close(); + } + + for (int i = 0; i < numSessions; i++) + { + SimpleString subName = new SimpleString(threadNum + "sub" + i); + + s.deleteQueue(subName); + } + + s.close(); + + long end = System.currentTimeMillis(); + + log.info("duration " + (end - start)); + } + + protected void doTestD(final ClientSessionFactory sf, final int threadNum) throws Exception + { + long start = System.currentTimeMillis(); + + ClientSession s = sf.createSession(false, false, false); + + final int numMessages = 100; + + final int numSessions = 10; + + Set<ClientConsumer> consumers = new HashSet<ClientConsumer>(); + Set<ClientSession> sessions = new HashSet<ClientSession>(); + + for (int i = 0; i < numSessions; i++) + { + SimpleString subName = new SimpleString(threadNum + " sub" + i); + + ClientSession sessConsume = sf.createSession(false, false, false); + + sessConsume.createQueue(ADDRESS, subName, null, false); + + ClientConsumer consumer = sessConsume.createConsumer(subName); + + consumers.add(consumer); + + sessions.add(sessConsume); + } + + ClientSession sessSend = sf.createSession(false, false, false); + + ClientProducer producer = sessSend.createProducer(ADDRESS); + + sendMessages(sessSend, producer, numMessages, threadNum); + + sessSend.rollback(); + + sendMessages(sessSend, producer, numMessages, threadNum); + + sessSend.commit(); + + for (ClientSession session : sessions) + { + session.start(); + } + + Set<MyHandler> handlers = new HashSet<MyHandler>(); + + for (ClientConsumer consumer : consumers) + { + MyHandler handler = new MyHandler(threadNum, numMessages); + + consumer.setMessageHandler(handler); + + handlers.add(handler); + } + + for (MyHandler handler : handlers) + { + boolean ok = handler.latch.await(LATCH_WAIT, TimeUnit.MILLISECONDS); + + if (!ok) + { + throw new Exception("Timed out waiting for messages on handler " + System.identityHashCode(handler) + + " threadnum " + + threadNum); + } + + if (handler.failure != null) + { + throw new Exception("Handler failed: " + handler.failure); + } + } + + handlers.clear(); + + // Set handlers to null + for (ClientConsumer consumer : consumers) + { + consumer.setMessageHandler(null); + } + + for (ClientSession session : sessions) + { + session.rollback(); + } + + // New handlers + for (ClientConsumer consumer : consumers) + { + MyHandler handler = new MyHandler(threadNum, numMessages); + + consumer.setMessageHandler(handler); + + handlers.add(handler); + } + + for (MyHandler handler : handlers) + { + boolean ok = handler.latch.await(LATCH_WAIT, TimeUnit.MILLISECONDS); + + if (!ok) + { + throw new Exception("Timed out waiting for messages on handler " + System.identityHashCode(handler) + + " threadnum " + + threadNum); + } + + if (handler.failure != null) + { + throw new Exception("Handler failed on rollback: " + handler.failure); + } + } + + for (ClientSession session : sessions) + { + session.commit(); + } + + sessSend.close(); + for (ClientSession session : sessions) + { + session.close(); + } + + for (int i = 0; i < numSessions; i++) + { + SimpleString subName = new SimpleString(threadNum + " sub" + i); + + s.deleteQueue(subName); + } + + s.close(); + + long end = System.currentTimeMillis(); + + log.info("duration " + (end - start)); + } + + // Now with synchronous receive() + + protected void doTestE(final ClientSessionFactory sf, final int threadNum) throws Exception + { + long start = System.currentTimeMillis(); + + ClientSession s = sf.createSession(false, false, false); + + final int numMessages = 100; + + final int numSessions = 10; + + Set<ClientConsumer> consumers = new HashSet<ClientConsumer>(); + Set<ClientSession> sessions = new HashSet<ClientSession>(); + + for (int i = 0; i < numSessions; i++) + { + SimpleString subName = new SimpleString(threadNum + "sub" + i); + + ClientSession sessConsume = sf.createSession(false, true, true); + + sessConsume.start(); + + sessConsume.createQueue(ADDRESS, subName, null, false); + + ClientConsumer consumer = sessConsume.createConsumer(subName); + + consumers.add(consumer); + + sessions.add(sessConsume); + } + + ClientSession sessSend = sf.createSession(false, true, true); + + ClientProducer producer = sessSend.createProducer(ADDRESS); + + sendMessages(sessSend, producer, numMessages, threadNum); + + consumeMessages(consumers, numMessages, threadNum); + + sessSend.close(); + for (ClientSession session : sessions) + { + session.close(); + } + + for (int i = 0; i < numSessions; i++) + { + SimpleString subName = new SimpleString(threadNum + "sub" + i); + + s.deleteQueue(subName); + } + + s.close(); + + long end = System.currentTimeMillis(); + + log.info("duration " + (end - start)); + } + + protected void doTestF(final ClientSessionFactory sf, final int threadNum) throws Exception + { + long start = System.currentTimeMillis(); + + ClientSession s = sf.createSession(false, false, false); + + final int numMessages = 100; + + final int numSessions = 10; + + Set<ClientConsumer> consumers = new HashSet<ClientConsumer>(); + Set<ClientSession> sessions = new HashSet<ClientSession>(); + + for (int i = 0; i < numSessions; i++) + { + SimpleString subName = new SimpleString(threadNum + "sub" + i); + + ClientSession sessConsume = sf.createSession(false, true, true); + + sessConsume.createQueue(ADDRESS, subName, null, false); + + ClientConsumer consumer = sessConsume.createConsumer(subName); + + consumers.add(consumer); + + sessions.add(sessConsume); + } + + ClientSession sessSend = sf.createSession(false, true, true); + + ClientProducer producer = sessSend.createProducer(ADDRESS); + + sendMessages(sessSend, producer, numMessages, threadNum); + + for (ClientSession session : sessions) + { + session.start(); + } + + consumeMessages(consumers, numMessages, threadNum); + + sessSend.close(); + for (ClientSession session : sessions) + { + session.close(); + } + + for (int i = 0; i < numSessions; i++) + { + SimpleString subName = new SimpleString(threadNum + "sub" + i); + + s.deleteQueue(subName); + } + + s.close(); + + long end = System.currentTimeMillis(); + + log.info("duration " + (end - start)); + } + + protected void doTestG(final ClientSessionFactory sf, final int threadNum) throws Exception + { + long start = System.currentTimeMillis(); + + ClientSession s = sf.createSession(false, false, false); + + final int numMessages = 100; + + final int numSessions = 10; + + Set<ClientConsumer> consumers = new HashSet<ClientConsumer>(); + Set<ClientSession> sessions = new HashSet<ClientSession>(); + + for (int i = 0; i < numSessions; i++) + { + SimpleString subName = new SimpleString(threadNum + "sub" + i); + + ClientSession sessConsume = sf.createSession(false, false, false); + + sessConsume.start(); + + sessConsume.createQueue(ADDRESS, subName, null, false); + + ClientConsumer consumer = sessConsume.createConsumer(subName); + + consumers.add(consumer); + + sessions.add(sessConsume); + } + + ClientSession sessSend = sf.createSession(false, false, false); + + ClientProducer producer = sessSend.createProducer(ADDRESS); + + sendMessages(sessSend, producer, numMessages, threadNum); + + sessSend.rollback(); + + sendMessages(sessSend, producer, numMessages, threadNum); + + sessSend.commit(); + + consumeMessages(consumers, numMessages, threadNum); + + for (ClientSession session : sessions) + { + session.rollback(); + } + + consumeMessages(consumers, numMessages, threadNum); + + for (ClientSession session : sessions) + { + session.commit(); + } + + sessSend.close(); + for (ClientSession session : sessions) + { + session.close(); + } + + for (int i = 0; i < numSessions; i++) + { + SimpleString subName = new SimpleString(threadNum + "sub" + i); + + s.deleteQueue(subName); + } + + s.close(); + + long end = System.currentTimeMillis(); + + log.info("duration " + (end - start)); + } + + protected void doTestH(final ClientSessionFactory sf, final int threadNum) throws Exception + { + long start = System.currentTimeMillis(); + + ClientSession s = sf.createSession(false, false, false); + + final int numMessages = 100; + + final int numSessions = 10; + + Set<ClientConsumer> consumers = new HashSet<ClientConsumer>(); + Set<ClientSession> sessions = new HashSet<ClientSession>(); + + for (int i = 0; i < numSessions; i++) + { + SimpleString subName = new SimpleString(threadNum + "sub" + i); + + ClientSession sessConsume = sf.createSession(false, false, false); + + sessConsume.createQueue(ADDRESS, subName, null, false); + + ClientConsumer consumer = sessConsume.createConsumer(subName); + + consumers.add(consumer); + + sessions.add(sessConsume); + } + + ClientSession sessSend = sf.createSession(false, false, false); + + ClientProducer producer = sessSend.createProducer(ADDRESS); + + sendMessages(sessSend, producer, numMessages, threadNum); + + sessSend.rollback(); + + sendMessages(sessSend, producer, numMessages, threadNum); + + sessSend.commit(); + + for (ClientSession session : sessions) + { + session.start(); + } + + consumeMessages(consumers, numMessages, threadNum); + + for (ClientSession session : sessions) + { + session.rollback(); + } + + consumeMessages(consumers, numMessages, threadNum); + + for (ClientSession session : sessions) + { + session.commit(); + } + + sessSend.close(); + for (ClientSession session : sessions) + { + session.close(); + } + + for (int i = 0; i < numSessions; i++) + { + SimpleString subName = new SimpleString(threadNum + "sub" + i); + + s.deleteQueue(subName); + } + + s.close(); + + long end = System.currentTimeMillis(); + + log.info("duration " + (end - start)); + } + + protected void doTestI(final ClientSessionFactory sf, final int threadNum) throws Exception + { + ClientSession sessCreate = sf.createSession(false, true, true); + + sessCreate.createQueue(ADDRESS, new SimpleString(threadNum + ADDRESS.toString()), null, false); + + ClientSession sess = sf.createSession(false, true, true); + + sess.start(); + + ClientConsumer consumer = sess.createConsumer(new SimpleString(threadNum + ADDRESS.toString())); + + ClientProducer producer = sess.createProducer(ADDRESS); + + ClientMessage message = sess.createClientMessage(HornetQTextMessage.TYPE, + false, + 0, + System.currentTimeMillis(), + (byte)1); + producer.send(message); + + ClientMessage message2 = consumer.receive(RECEIVE_TIMEOUT); + + assertNotNull(message2); + + message2.acknowledge(); + + sess.close(); + + sessCreate.deleteQueue(new SimpleString(threadNum + ADDRESS.toString())); + + sessCreate.close(); + } + + protected void doTestJ(final ClientSessionFactory sf, final int threadNum) throws Exception + { + ClientSession sessCreate = sf.createSession(false, true, true); + + sessCreate.createQueue(ADDRESS, new SimpleString(threadNum + ADDRESS.toString()), null, false); + + ClientSession sess = sf.createSession(false, true, true); + + sess.start(); + + ClientConsumer consumer = sess.createConsumer(new SimpleString(threadNum + ADDRESS.toString())); + + ClientProducer producer = sess.createProducer(ADDRESS); + + ClientMessage message = sess.createClientMessage(HornetQTextMessage.TYPE, + false, + 0, + System.currentTimeMillis(), + (byte)1); + producer.send(message); + + ClientMessage message2 = consumer.receive(RECEIVE_TIMEOUT); + + assertNotNull(message2); + + message2.acknowledge(); + + sess.close(); + + sessCreate.deleteQueue(new SimpleString(threadNum + ADDRESS.toString())); + + sessCreate.close(); + } + + protected void doTestK(final ClientSessionFactory sf, final int threadNum) throws Exception + { + ClientSession s = sf.createSession(false, false, false); + + s.createQueue(ADDRESS, new SimpleString(threadNum + ADDRESS.toString()), null, false); + + final int numConsumers = 100; + + for (int i = 0; i < numConsumers; i++) + { + ClientConsumer consumer = s.createConsumer(new SimpleString(threadNum + ADDRESS.toString())); + + consumer.close(); + } + + s.deleteQueue(new SimpleString(threadNum + ADDRESS.toString())); + + s.close(); + } + + /* + * This test tests failure during create connection + */ + protected void doTestL(final ClientSessionFactory sf) throws Exception + { + final int numSessions = 10; + + for (int i = 0; i < numSessions; i++) + { + ClientSession session = sf.createSession(false, false, false); + + session.close(); + } + } + + // Browsers + // FIXME - this test won't work until we use a proper iterator for browsing a queue. + // Making a copy of the queue for a browser consumer doesn't work well with replication since + // When replicating the create consumer (browser) to the backup, when executed on the backup the + // backup may have different messages in its queue since been added on different threads. + // So when replicating deliveries they may not be found. + // https://jira.jboss.org/jira/browse/JBMESSAGING-1433 + // protected void doTestM(final ClientSessionFactory sf, final int threadNum) throws Exception + // { + // long start = System.currentTimeMillis(); + // + // ClientSession sessSend = sf.createSession(false, true, true, false); + // + // ClientSession sessConsume = sf.createSession(false, true, true, false); + // + // sessConsume.createQueue(ADDRESS, new SimpleString(threadNum + "sub"), null, false, false); + // + // final int numMessages = 100; + // + // ClientProducer producer = sessSend.createProducer(ADDRESS); + // + // sendMessages(sessSend, producer, numMessages, threadNum); + // + // ClientConsumer browser = sessConsume.createConsumer(new SimpleString(threadNum + "sub"), + // null, false, true); + // + // Map<Integer, Integer> consumerCounts = new HashMap<Integer, Integer>(); + // + // for (int i = 0; i < numMessages; i++) + // { + // ClientMessage msg = browser.receive(RECEIVE_TIMEOUT); + // + // assertNotNull(msg); + // + // int tn = (Integer)msg.getProperty(new SimpleString("threadnum")); + // int cnt = (Integer)msg.getProperty(new SimpleString("count")); + // + // Integer c = consumerCounts.get(tn); + // if (c == null) + // { + // c = new Integer(cnt); + // } + // + // if (cnt != c.intValue()) + // { + // throw new Exception("Invalid count, expected " + c + " got " + cnt); + // } + // + // c++; + // + // //Wrap + // if (c == numMessages) + // { + // c = 0; + // } + // + // consumerCounts.put(tn, c); + // + // msg.acknowledge(); + // } + // + // sessConsume.close(); + // + // sessConsume = sf.createSession(false, true, true, false); + // + // browser = sessConsume.createConsumer(new SimpleString(threadNum + "sub"), + // null, false, true); + // + // //Messages should still be there + // + // consumerCounts.clear(); + // + // for (int i = 0; i < numMessages; i++) + // { + // ClientMessage msg = browser.receive(RECEIVE_TIMEOUT); + // + // assertNotNull(msg); + // + // int tn = (Integer)msg.getProperty(new SimpleString("threadnum")); + // int cnt = (Integer)msg.getProperty(new SimpleString("count")); + // + // Integer c = consumerCounts.get(tn); + // if (c == null) + // { + // c = new Integer(cnt); + // } + // + // if (cnt != c.intValue()) + // { + // throw new Exception("Invalid count, expected " + c + " got " + cnt); + // } + // + // c++; + // + // //Wrap + // if (c == numMessages) + // { + // c = 0; + // } + // + // consumerCounts.put(tn, c); + // + // msg.acknowledge(); + // } + // + // sessConsume.close(); + // + // sessSend.deleteQueue(new SimpleString(threadNum + "sub")); + // + // sessSend.close(); + // + // long end = System.currentTimeMillis(); + // + // log.info("duration " + (end - start)); + // } + + protected void doTestN(final ClientSessionFactory sf, final int threadNum) throws Exception + { + ClientSession sessCreate = sf.createSession(false, true, true); + + sessCreate.createQueue(ADDRESS, new SimpleString(threadNum + ADDRESS.toString()), null, false); + + ClientSession sess = sf.createSession(false, true, true); + + sess.stop(); + + sess.start(); + + sess.stop(); + + ClientConsumer consumer = sess.createConsumer(new SimpleString(threadNum + ADDRESS.toString())); + + ClientProducer producer = sess.createProducer(ADDRESS); + + ClientMessage message = sess.createClientMessage(HornetQTextMessage.TYPE, + false, + 0, + System.currentTimeMillis(), + (byte)1); + producer.send(message); + + sess.start(); + + ClientMessage message2 = consumer.receive(RECEIVE_TIMEOUT); + + assertNotNull(message2); + + message2.acknowledge(); + + sess.stop(); + + sess.start(); + + sess.close(); + + sessCreate.deleteQueue(new SimpleString(threadNum + ADDRESS.toString())); + + sessCreate.close(); + } + + protected int getLatchWait() + { + return 60000; + } + + protected int getNumIterations() + { + return 3; + } + + @Override + protected void setUp() throws Exception + { + super.setUp(); + + log.info("************ Starting test " + getName()); + } + + @Override + protected void tearDown() throws Exception + { + if (liveServer != null && liveServer.isStarted()) + { + liveServer.stop(); + } + + liveServer = null; + + super.tearDown(); + } + + // Private ------------------------------------------------------- + + private void runTestMultipleThreads(final RunnableT runnable, + final int numThreads, + final boolean failOnCreateConnection) throws Exception + { + runTestMultipleThreads(runnable, numThreads, failOnCreateConnection, 1000); + } + + private void runTestMultipleThreads(final RunnableT runnable, + final int numThreads, + final boolean failOnCreateConnection, + final long failDelay) throws Exception + { + + runMultipleThreadsFailoverTest(runnable, numThreads, getNumIterations(), failOnCreateConnection, failDelay); + } + + /** + * @return + */ + protected ClientSessionFactoryInternal createSessionFactory() + { + final ClientSessionFactoryInternal sf = new ClientSessionFactoryImpl(new TransportConfiguration("org.hornetq.core.remoting.impl.invm.InVMConnectorFactory")); + sf.setReconnectAttempts(-1); + sf.setUseReattach(true); + + return sf; + } + + protected void stop() throws Exception + { + liveServer.stop(); + + System.gc(); + + assertEquals(0, InVMRegistry.instance.size()); + } + + private void sendMessages(final ClientSession sessSend, + final ClientProducer producer, + final int numMessages, + final int threadNum) throws Exception + { + for (int i = 0; i < numMessages; i++) + { + ClientMessage message = sessSend.createClientMessage(HornetQBytesMessage.TYPE, + false, + 0, + System.currentTimeMillis(), + (byte)1); + message.putIntProperty(new SimpleString("threadnum"), threadNum); + message.putIntProperty(new SimpleString("count"), i); + setBody(message); + producer.send(message); + } + } + + private void consumeMessages(final Set<ClientConsumer> consumers, final int numMessages, final int threadNum) throws Exception + { + // We make sure the messages arrive in the order they were sent from a particular producer + Map<ClientConsumer, Map<Integer, Integer>> counts = new HashMap<ClientConsumer, Map<Integer, Integer>>(); + + for (int i = 0; i < numMessages; i++) + { + for (ClientConsumer consumer : consumers) + { + Map<Integer, Integer> consumerCounts = counts.get(consumer); + + if (consumerCounts == null) + { + consumerCounts = new HashMap<Integer, Integer>(); + counts.put(consumer, consumerCounts); + } + + ClientMessage msg = consumer.receive(RECEIVE_TIMEOUT); + + assertNotNull(msg); + + int tn = (Integer)msg.getProperty(new SimpleString("threadnum")); + int cnt = (Integer)msg.getProperty(new SimpleString("count")); + + Integer c = consumerCounts.get(tn); + if (c == null) + { + c = new Integer(cnt); + } + + if (tn == threadNum && cnt != c.intValue()) + { + throw new Exception("Invalid count, expected " + tn + ": " + c + " got " + cnt); + } + + c++; + + // Wrap + if (c == numMessages) + { + c = 0; + } + + consumerCounts.put(tn, c); + + msg.acknowledge(); + } + } + } + + // Inner classes ------------------------------------------------- + + private class MyHandler implements MessageHandler + { + CountDownLatch latch = new CountDownLatch(1); + + private final Map<Integer, Integer> counts = new HashMap<Integer, Integer>(); + + volatile String failure; + + final int tn; + + final int numMessages; + + volatile boolean done; + + synchronized void reset() + { + counts.clear(); + + done = false; + + failure = null; + + latch = new CountDownLatch(1); + } + + MyHandler(final int threadNum, final int numMessages) + { + tn = threadNum; + + this.numMessages = numMessages; + } + + public synchronized void onMessage(final ClientMessage message) + { + try + { + message.acknowledge(); + } + catch (HornetQException me) + { + log.error("Failed to process", me); + } + + if (done) + { + return; + } + + int threadNum = (Integer)message.getProperty(new SimpleString("threadnum")); + int cnt = (Integer)message.getProperty(new SimpleString("count")); + + Integer c = counts.get(threadNum); + if (c == null) + { + c = new Integer(cnt); + } + + if (tn == threadNum && cnt != c.intValue()) + { + failure = "Invalid count, expected " + threadNum + ":" + c + " got " + cnt; + log.error(failure); + + latch.countDown(); + } + + if (!checkSize(message)) + { + failure = "Invalid size on message"; + log.error(failure); + latch.countDown(); + } + + if (tn == threadNum && c == numMessages - 1) + { + done = true; + latch.countDown(); + } + + c++; + // Wrap around at numMessages + if (c == numMessages) + { + c = 0; + } + + counts.put(threadNum, c); + + } + } +} \ No newline at end of file Copied: branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/integration/cluster/reattach/MultiThreadReattachSupport.java (from rev 7962, branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/integration/cluster/failover/MultiThreadFailoverSupport.java) =================================================================== --- branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/integration/cluster/reattach/MultiThreadReattachSupport.java (rev 0) +++ branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/integration/cluster/reattach/MultiThreadReattachSupport.java 2009-09-22 15:08:41 UTC (rev 7979) @@ -0,0 +1,278 @@ +/* + * Copyright 2009 Red Hat, Inc. + * Red Hat licenses this file to you under the Apache License, version + * 2.0 (the "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * http://www.apache.org/licenses/LICENSE-2.0 + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or + * implied. See the License for the specific language governing + * permissions and limitations under the License. + */ + + +package org.hornetq.tests.integration.cluster.reattach; + +import java.util.ArrayList; +import java.util.List; +import java.util.Timer; +import java.util.TimerTask; + +import org.hornetq.core.client.ClientSession; +import org.hornetq.core.client.ClientSessionFactory; +import org.hornetq.core.client.impl.ClientSessionFactoryInternal; +import org.hornetq.core.client.impl.ClientSessionInternal; +import org.hornetq.core.exception.HornetQException; +import org.hornetq.core.logging.Logger; +import org.hornetq.core.remoting.impl.RemotingConnectionImpl; +import org.hornetq.core.remoting.impl.invm.InVMConnector; +import org.hornetq.tests.util.ServiceTestBase; + +/** + * A MultiThreadFailoverSupport + * + * @author <a href="mailto:time.fox@jboss.org">Tim Fox</a> + * @author <a href="mailto:clebert.suconic@jboss.org">Clebert Suconic</a> + * + * Created Mar 17, 2009 11:15:02 AM + * + * + */ +public abstract class MultiThreadReattachSupport extends ServiceTestBase +{ + + // Constants ----------------------------------------------------- + + private final Logger log = Logger.getLogger(this.getClass()); + + // Attributes ---------------------------------------------------- + + private Timer timer; + + // Static -------------------------------------------------------- + + // Constructors -------------------------------------------------- + + // Public -------------------------------------------------------- + + // Package protected --------------------------------------------- + + // Protected ----------------------------------------------------- + + protected abstract void start() throws Exception; + + protected abstract void stop() throws Exception; + + protected abstract ClientSessionFactoryInternal createSessionFactory(); + + protected void setUp() throws Exception + { + super.setUp(); + timer = new Timer(); + } + + protected void tearDown() throws Exception + { + timer.cancel(); + timer = null; + super.tearDown(); + } + + protected boolean shouldFail() + { + return true; + } + + protected void runMultipleThreadsFailoverTest(final RunnableT runnable, + final int numThreads, + final int numIts, + final boolean failOnCreateConnection, + final long failDelay) throws Exception + { + for (int its = 0; its < numIts; its++) + { + log.info("Beginning iteration " + its); + + start(); + + final ClientSessionFactoryInternal sf = createSessionFactory(); + + final ClientSession session = sf.createSession(false, true, true); + + Failer failer = startFailer(failDelay, session, failOnCreateConnection); + + class Runner extends Thread + { + private volatile Throwable throwable; + + private final RunnableT test; + + private final int threadNum; + + Runner(final RunnableT test, final int threadNum) + { + this.test = test; + + this.threadNum = threadNum; + } + + @Override + public void run() + { + try + { + test.run(sf, threadNum); + } + catch (Throwable t) + { + throwable = t; + + log.error("Failed to run test", t); + + // Case a failure happened here, it should print the Thread dump + // Sending it to System.out, as it would show on the Tests report + System.out.println(threadDump(" - fired by MultiThreadRandomReattachTestBase::runTestMultipleThreads (" + t.getLocalizedMessage() + ")")); + } + } + } + + do + { + List<Runner> threads = new ArrayList<Runner>(); + + for (int i = 0; i < numThreads; i++) + { + Runner runner = new Runner(runnable, i); + + threads.add(runner); + + runner.start(); + } + + for (Runner thread : threads) + { + thread.join(); + + if (thread.throwable != null) + { + throw new Exception("Exception on thread " + thread, thread.throwable); + } + } + + runnable.checkFail(); + + } + while (!failer.isExecuted()); + + InVMConnector.resetFailures(); + + session.close(); + + assertEquals(0, sf.numSessions()); + + assertEquals(0, sf.numConnections()); + + sf.close(); + + stop(); + } + } + + + // Private ------------------------------------------------------- + + private Failer startFailer(final long time, final ClientSession session, final boolean failOnCreateConnection) + { + Failer failer = new Failer(session, failOnCreateConnection); + + // This is useful for debugging.. just change shouldFail to return false, and Failer will not be executed + if (shouldFail()) + { + timer.schedule(failer, (long)(time * Math.random()), 100); + } + + return failer; + } + + + // Inner classes ------------------------------------------------- + + + protected abstract class RunnableT extends Thread + { + private volatile String failReason; + + private volatile Throwable throwable; + + public void setFailed(final String reason, final Throwable throwable) + { + failReason = reason; + this.throwable = throwable; + } + + public void checkFail() + { + if (throwable != null) + { + log.error("Test failed: " + failReason, throwable); + } + if (failReason != null) + { + fail(failReason); + } + } + + public abstract void run(final ClientSessionFactory sf, final int threadNum) throws Exception; + } + + + + private class Failer extends TimerTask + { + private final ClientSession session; + + private boolean executed; + + private final boolean failOnCreateConnection; + + public Failer(final ClientSession session, final boolean failOnCreateConnection) + { + this.session = session; + + this.failOnCreateConnection = failOnCreateConnection; + } + + @Override + public synchronized void run() + { + log.info("** Failing connection"); + + RemotingConnectionImpl conn = (RemotingConnectionImpl)((ClientSessionInternal)session).getConnection(); + + if (failOnCreateConnection) + { + InVMConnector.numberOfFailures = 1; + InVMConnector.failOnCreateConnection = true; + } + else + { + conn.fail(new HornetQException(HornetQException.NOT_CONNECTED, "blah")); + } + + log.info("** Fail complete"); + + cancel(); + + executed = true; + } + + public synchronized boolean isExecuted() + { + return executed; + } + } + + + +} Copied: branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/integration/cluster/reattach/NettyMultiThreadRandomReattachTest.java (from rev 7962, branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/integration/cluster/failover/NettyMultiThreadRandomReattachTest.java) =================================================================== --- branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/integration/cluster/reattach/NettyMultiThreadRandomReattachTest.java (rev 0) +++ branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/integration/cluster/reattach/NettyMultiThreadRandomReattachTest.java 2009-09-22 15:08:41 UTC (rev 7979) @@ -0,0 +1,56 @@ +/* + * Copyright 2009 Red Hat, Inc. + * Red Hat licenses this file to you under the Apache License, version + * 2.0 (the "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * http://www.apache.org/licenses/LICENSE-2.0 + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or + * implied. See the License for the specific language governing + * permissions and limitations under the License. + */ + +package org.hornetq.tests.integration.cluster.reattach; + +import org.hornetq.core.client.impl.ClientSessionFactoryImpl; +import org.hornetq.core.client.impl.ClientSessionFactoryInternal; +import org.hornetq.core.config.Configuration; +import org.hornetq.core.config.TransportConfiguration; +import org.hornetq.core.config.impl.ConfigurationImpl; +import org.hornetq.core.server.HornetQ; + +/** + * A NettyMultiThreadRandomReattachTest + * + * @author <a href="mailto:tim.fox@jboss.com">Tim Fox</a> + * + * Created 18 Feb 2009 08:01:20 + * + * + */ +public class NettyMultiThreadRandomReattachTest extends MultiThreadRandomReattachTest +{ + @Override + protected void start() throws Exception + { + Configuration liveConf = new ConfigurationImpl(); + liveConf.setJMXManagementEnabled(false); + liveConf.setSecurityEnabled(false); + liveConf.getAcceptorConfigurations().clear(); + liveConf.getAcceptorConfigurations() + .add(new TransportConfiguration("org.hornetq.integration.transports.netty.NettyAcceptorFactory")); + liveServer = HornetQ.newHornetQServer(liveConf, false); + liveServer.start(); + } + + @Override + protected ClientSessionFactoryInternal createSessionFactory() + { + final ClientSessionFactoryInternal sf = new ClientSessionFactoryImpl(new TransportConfiguration("org.hornetq.integration.transports.netty.NettyConnectorFactory")); + sf.setUseReattach(true); + sf.setReconnectAttempts(-1); + return sf; + } + +} Copied: branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/integration/cluster/reattach/RandomReattachTest.java (from rev 7962, branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/integration/cluster/failover/RandomReattachTest.java) =================================================================== --- branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/integration/cluster/reattach/RandomReattachTest.java (rev 0) +++ branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/integration/cluster/reattach/RandomReattachTest.java 2009-09-22 15:08:41 UTC (rev 7979) @@ -0,0 +1,1481 @@ +/* + * Copyright 2009 Red Hat, Inc. + * Red Hat licenses this file to you under the Apache License, version + * 2.0 (the "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * http://www.apache.org/licenses/LICENSE-2.0 + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or + * implied. See the License for the specific language governing + * permissions and limitations under the License. + */ + +package org.hornetq.tests.integration.cluster.reattach; + +import java.util.HashSet; +import java.util.Set; +import java.util.Timer; +import java.util.TimerTask; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; + +import org.hornetq.core.client.ClientConsumer; +import org.hornetq.core.client.ClientMessage; +import org.hornetq.core.client.ClientProducer; +import org.hornetq.core.client.ClientSession; +import org.hornetq.core.client.ClientSessionFactory; +import org.hornetq.core.client.MessageHandler; +import org.hornetq.core.client.impl.ClientSessionFactoryImpl; +import org.hornetq.core.client.impl.ClientSessionInternal; +import org.hornetq.core.config.Configuration; +import org.hornetq.core.config.TransportConfiguration; +import org.hornetq.core.config.impl.ConfigurationImpl; +import org.hornetq.core.exception.HornetQException; +import org.hornetq.core.logging.Logger; +import org.hornetq.core.remoting.impl.invm.InVMRegistry; +import org.hornetq.core.server.HornetQ; +import org.hornetq.core.server.HornetQServer; +import org.hornetq.jms.client.HornetQTextMessage; +import org.hornetq.tests.util.UnitTestCase; +import org.hornetq.utils.SimpleString; + +/** + * A RandomFailoverSoakTest + * + * @author <a href="mailto:tim.fox@jboss.com">Tim Fox</a> + */ +public class RandomReattachTest extends UnitTestCase +{ + private static final Logger log = Logger.getLogger(RandomReattachTest.class); + + // Constants ----------------------------------------------------- + + private static final int RECEIVE_TIMEOUT = 10000; + + // Attributes ---------------------------------------------------- + + private static final SimpleString ADDRESS = new SimpleString("FailoverTestAddress"); + + private HornetQServer liveService; + + private Timer timer; + + // Static -------------------------------------------------------- + + // Constructors -------------------------------------------------- + + // Public -------------------------------------------------------- + + public void testA() throws Exception + { + runTest(new RunnableT() + { + public void run(final ClientSessionFactory sf) throws Exception + { + doTestA(sf); + } + }); + } + + public void testB() throws Exception + { + runTest(new RunnableT() + { + public void run(final ClientSessionFactory sf) throws Exception + { + doTestB(sf); + } + }); + } + + public void testC() throws Exception + { + runTest(new RunnableT() + { + public void run(final ClientSessionFactory sf) throws Exception + { + doTestC(sf); + } + }); + } + + public void testD() throws Exception + { + runTest(new RunnableT() + { + public void run(final ClientSessionFactory sf) throws Exception + { + doTestD(sf); + } + }); + } + + public void testE() throws Exception + { + runTest(new RunnableT() + { + public void run(final ClientSessionFactory sf) throws Exception + { + doTestE(sf); + } + }); + } + + public void testF() throws Exception + { + runTest(new RunnableT() + { + public void run(final ClientSessionFactory sf) throws Exception + { + doTestF(sf); + } + }); + } + + public void testG() throws Exception + { + runTest(new RunnableT() + { + public void run(final ClientSessionFactory sf) throws Exception + { + doTestG(sf); + } + }); + } + + public void testH() throws Exception + { + runTest(new RunnableT() + { + public void run(final ClientSessionFactory sf) throws Exception + { + doTestH(sf); + } + }); + } + + public void testI() throws Exception + { + runTest(new RunnableT() + { + public void run(final ClientSessionFactory sf) throws Exception + { + doTestI(sf); + } + }); + } + + public void testJ() throws Exception + { + runTest(new RunnableT() + { + public void run(final ClientSessionFactory sf) throws Exception + { + doTestJ(sf); + } + }); + } + + public void testK() throws Exception + { + runTest(new RunnableT() + { + public void run(final ClientSessionFactory sf) throws Exception + { + doTestK(sf); + } + }); + } + + public void testL() throws Exception + { + runTest(new RunnableT() + { + public void run(final ClientSessionFactory sf) throws Exception + { + doTestL(sf); + } + }); + } + + public void testN() throws Exception + { + runTest(new RunnableT() + { + public void run(final ClientSessionFactory sf) throws Exception + { + doTestN(sf); + } + }); + } + + public void runTest(final RunnableT runnable) throws Exception + { + final int numIts = getNumIterations(); + + for (int its = 0; its < numIts; its++) + { + start(); + + ClientSessionFactoryImpl sf = new ClientSessionFactoryImpl(new TransportConfiguration("org.hornetq.core.remoting.impl.invm.InVMConnectorFactory")); + + sf.setReconnectAttempts(-1); + + sf.setUseReattach(true); + + ClientSession session = sf.createSession(false, false, false); + + Failer failer = startFailer(1000, session); + + do + { + runnable.run(sf); + } + while (!failer.isExecuted()); + + session.close(); + + assertEquals(0, sf.numSessions()); + + assertEquals(0, sf.numConnections()); + + stop(); + } + } + + // Package protected --------------------------------------------- + + // Protected ----------------------------------------------------- + + protected void doTestA(final ClientSessionFactory sf) throws Exception + { + long start = System.currentTimeMillis(); + + ClientSession s = sf.createSession(false, false, false); + + final int numMessages = 100; + + final int numSessions = 10; + + Set<ClientConsumer> consumers = new HashSet<ClientConsumer>(); + Set<ClientSession> sessions = new HashSet<ClientSession>(); + + for (int i = 0; i < numSessions; i++) + { + SimpleString subName = new SimpleString("sub" + i); + + ClientSession sessConsume = sf.createSession(false, true, true); + + sessConsume.start(); + + sessConsume.createQueue(ADDRESS, subName, null, false); + + ClientConsumer consumer = sessConsume.createConsumer(subName); + + consumers.add(consumer); + + sessions.add(sessConsume); + } + + ClientSession sessSend = sf.createSession(false, true, true); + + ClientProducer producer = sessSend.createProducer(ADDRESS); + + for (int i = 0; i < numMessages; i++) + { + ClientMessage message = sessSend.createClientMessage(HornetQTextMessage.TYPE, + false, + 0, + System.currentTimeMillis(), + (byte)1); + message.putIntProperty(new SimpleString("count"), i); + producer.send(message); + } + + class MyHandler implements MessageHandler + { + final CountDownLatch latch = new CountDownLatch(1); + + volatile int count; + + public void onMessage(ClientMessage message) + { + if (count == numMessages) + { + fail("Too many messages"); + } + + assertEquals(count, message.getProperty(new SimpleString("count"))); + + count++; + + try + { + message.acknowledge(); + } + catch (HornetQException me) + { + log.error("Failed to process", me); + } + + if (count == numMessages) + { + latch.countDown(); + } + } + } + + Set<MyHandler> handlers = new HashSet<MyHandler>(); + + for (ClientConsumer consumer : consumers) + { + MyHandler handler = new MyHandler(); + + consumer.setMessageHandler(handler); + + handlers.add(handler); + } + + for (MyHandler handler : handlers) + { + boolean ok = handler.latch.await(5000, TimeUnit.MILLISECONDS); + + assertTrue("Didn't receive all messages", ok); + } + + sessSend.close(); + for (ClientSession session : sessions) + { + session.close(); + } + + for (int i = 0; i < numSessions; i++) + { + SimpleString subName = new SimpleString("sub" + i); + + s.deleteQueue(subName); + } + + s.close(); + + long end = System.currentTimeMillis(); + + log.info("duration " + (end - start)); + } + + protected void doTestB(final ClientSessionFactory sf) throws Exception + { + long start = System.currentTimeMillis(); + + ClientSession s = sf.createSession(false, false, false); + + final int numMessages = 100; + + final int numSessions = 50; + + Set<ClientConsumer> consumers = new HashSet<ClientConsumer>(); + Set<ClientSession> sessions = new HashSet<ClientSession>(); + + for (int i = 0; i < numSessions; i++) + { + SimpleString subName = new SimpleString("sub" + i); + + ClientSession sessConsume = sf.createSession(false, true, true); + + sessConsume.createQueue(ADDRESS, subName, null, false); + + ClientConsumer consumer = sessConsume.createConsumer(subName); + + consumers.add(consumer); + + sessions.add(sessConsume); + } + + ClientSession sessSend = sf.createSession(false, true, true); + + ClientProducer producer = sessSend.createProducer(ADDRESS); + + for (int i = 0; i < numMessages; i++) + { + ClientMessage message = sessSend.createClientMessage(HornetQTextMessage.TYPE, + false, + 0, + System.currentTimeMillis(), + (byte)1); + message.putIntProperty(new SimpleString("count"), i); + producer.send(message); + } + + for (ClientSession session : sessions) + { + session.start(); + } + + class MyHandler implements MessageHandler + { + final CountDownLatch latch = new CountDownLatch(1); + + volatile int count; + + public void onMessage(ClientMessage message) + { + if (count == numMessages) + { + fail("Too many messages"); + } + + assertEquals(count, message.getProperty(new SimpleString("count"))); + + count++; + + if (count == numMessages) + { + latch.countDown(); + } + } + } + + Set<MyHandler> handlers = new HashSet<MyHandler>(); + + for (ClientConsumer consumer : consumers) + { + MyHandler handler = new MyHandler(); + + consumer.setMessageHandler(handler); + + handlers.add(handler); + } + + for (MyHandler handler : handlers) + { + boolean ok = handler.latch.await(10000, TimeUnit.MILLISECONDS); + + assertTrue(ok); + } + + sessSend.close(); + + for (ClientSession session : sessions) + { + session.close(); + } + + for (int i = 0; i < numSessions; i++) + { + SimpleString subName = new SimpleString("sub" + i); + + s.deleteQueue(subName); + } + + s.close(); + + long end = System.currentTimeMillis(); + + log.info("duration " + (end - start)); + + } + + protected void doTestC(final ClientSessionFactory sf) throws Exception + { + long start = System.currentTimeMillis(); + + ClientSession s = sf.createSession(false, false, false); + + final int numMessages = 100; + + final int numSessions = 10; + + Set<ClientConsumer> consumers = new HashSet<ClientConsumer>(); + Set<ClientSession> sessions = new HashSet<ClientSession>(); + + for (int i = 0; i < numSessions; i++) + { + SimpleString subName = new SimpleString("sub" + i); + + ClientSession sessConsume = sf.createSession(false, false, false); + + sessConsume.start(); + + sessConsume.createQueue(ADDRESS, subName, null, false); + + ClientConsumer consumer = sessConsume.createConsumer(subName); + + consumers.add(consumer); + + sessions.add(sessConsume); + } + + ClientSession sessSend = sf.createSession(false, true, true); + + ClientProducer producer = sessSend.createProducer(ADDRESS); + + for (int i = 0; i < numMessages; i++) + { + ClientMessage message = sessSend.createClientMessage(HornetQTextMessage.TYPE, + false, + 0, + System.currentTimeMillis(), + (byte)1); + message.putIntProperty(new SimpleString("count"), i); + producer.send(message); + } + + sessSend.rollback(); + + for (int i = 0; i < numMessages; i++) + { + ClientMessage message = sessSend.createClientMessage(HornetQTextMessage.TYPE, + false, + 0, + System.currentTimeMillis(), + (byte)1); + message.putIntProperty(new SimpleString("count"), i); + producer.send(message); + } + + sessSend.commit(); + + class MyHandler implements MessageHandler + { + final CountDownLatch latch = new CountDownLatch(1); + + volatile int count; + + public void onMessage(ClientMessage message) + { + if (count == numMessages) + { + fail("Too many messages"); + } + + assertEquals(count, message.getProperty(new SimpleString("count"))); + + count++; + + if (count == numMessages) + { + latch.countDown(); + } + } + } + + Set<MyHandler> handlers = new HashSet<MyHandler>(); + + for (ClientConsumer consumer : consumers) + { + MyHandler handler = new MyHandler(); + + consumer.setMessageHandler(handler); + + handlers.add(handler); + } + + for (MyHandler handler : handlers) + { + boolean ok = handler.latch.await(10000, TimeUnit.MILLISECONDS); + + assertTrue(ok); + } + + handlers.clear(); + + // New handlers + for (ClientConsumer consumer : consumers) + { + MyHandler handler = new MyHandler(); + + consumer.setMessageHandler(handler); + + handlers.add(handler); + } + + for (ClientSession session : sessions) + { + session.rollback(); + } + + for (MyHandler handler : handlers) + { + boolean ok = handler.latch.await(10000, TimeUnit.MILLISECONDS); + + assertTrue(ok); + } + + for (ClientSession session : sessions) + { + session.commit(); + } + + sessSend.close(); + for (ClientSession session : sessions) + { + session.close(); + } + + for (int i = 0; i < numSessions; i++) + { + SimpleString subName = new SimpleString("sub" + i); + + s.deleteQueue(subName); + } + + s.close(); + + long end = System.currentTimeMillis(); + + log.info("duration " + (end - start)); + } + + protected void doTestD(final ClientSessionFactory sf) throws Exception + { + long start = System.currentTimeMillis(); + + ClientSession s = sf.createSession(false, false, false); + + final int numMessages = 100; + + final int numSessions = 10; + + Set<ClientConsumer> consumers = new HashSet<ClientConsumer>(); + Set<ClientSession> sessions = new HashSet<ClientSession>(); + + for (int i = 0; i < numSessions; i++) + { + SimpleString subName = new SimpleString("sub" + i); + + ClientSession sessConsume = sf.createSession(false, false, false); + + sessConsume.createQueue(ADDRESS, subName, null, false); + + ClientConsumer consumer = sessConsume.createConsumer(subName); + + consumers.add(consumer); + + sessions.add(sessConsume); + } + + ClientSession sessSend = sf.createSession(false, true, true); + + ClientProducer producer = sessSend.createProducer(ADDRESS); + + for (int i = 0; i < numMessages; i++) + { + ClientMessage message = sessSend.createClientMessage(HornetQTextMessage.TYPE, + false, + 0, + System.currentTimeMillis(), + (byte)1); + message.putIntProperty(new SimpleString("count"), i); + producer.send(message); + } + + sessSend.rollback(); + + for (int i = 0; i < numMessages; i++) + { + ClientMessage message = sessSend.createClientMessage(HornetQTextMessage.TYPE, + false, + 0, + System.currentTimeMillis(), + (byte)1); + message.putIntProperty(new SimpleString("count"), i); + producer.send(message); + } + + sessSend.commit(); + + for (ClientSession session : sessions) + { + session.start(); + } + + class MyHandler implements MessageHandler + { + final CountDownLatch latch = new CountDownLatch(1); + + volatile int count; + + public void onMessage(ClientMessage message) + { + if (count == numMessages) + { + fail("Too many messages"); + } + + assertEquals(count, message.getProperty(new SimpleString("count"))); + + count++; + + if (count == numMessages) + { + latch.countDown(); + } + } + } + + Set<MyHandler> handlers = new HashSet<MyHandler>(); + + for (ClientConsumer consumer : consumers) + { + MyHandler handler = new MyHandler(); + + consumer.setMessageHandler(handler); + + handlers.add(handler); + } + + for (MyHandler handler : handlers) + { + boolean ok = handler.latch.await(10000, TimeUnit.MILLISECONDS); + + assertTrue(ok); + } + + handlers.clear(); + + // New handlers + for (ClientConsumer consumer : consumers) + { + MyHandler handler = new MyHandler(); + + consumer.setMessageHandler(handler); + + handlers.add(handler); + } + + for (ClientSession session : sessions) + { + session.rollback(); + } + + for (MyHandler handler : handlers) + { + boolean ok = handler.latch.await(10000, TimeUnit.MILLISECONDS); + + assertTrue(ok); + } + + for (ClientSession session : sessions) + { + session.commit(); + } + + sessSend.close(); + for (ClientSession session : sessions) + { + session.close(); + } + + for (int i = 0; i < numSessions; i++) + { + SimpleString subName = new SimpleString("sub" + i); + + s.deleteQueue(subName); + } + + s.close(); + + long end = System.currentTimeMillis(); + + log.info("duration " + (end - start)); + } + + // Now with synchronous receive() + + protected void doTestE(final ClientSessionFactory sf) throws Exception + { + long start = System.currentTimeMillis(); + + ClientSession s = sf.createSession(false, false, false); + + final int numMessages = 100; + + final int numSessions = 10; + + Set<ClientConsumer> consumers = new HashSet<ClientConsumer>(); + Set<ClientSession> sessions = new HashSet<ClientSession>(); + + for (int i = 0; i < numSessions; i++) + { + SimpleString subName = new SimpleString("sub" + i); + + ClientSession sessConsume = sf.createSession(false, true, true); + + sessConsume.start(); + + sessConsume.createQueue(ADDRESS, subName, null, false); + + ClientConsumer consumer = sessConsume.createConsumer(subName); + + consumers.add(consumer); + + sessions.add(sessConsume); + } + + ClientSession sessSend = sf.createSession(false, true, true); + + ClientProducer producer = sessSend.createProducer(ADDRESS); + + for (int i = 0; i < numMessages; i++) + { + ClientMessage message = sessSend.createClientMessage(HornetQTextMessage.TYPE, + false, + 0, + System.currentTimeMillis(), + (byte)1); + message.putIntProperty(new SimpleString("count"), i); + producer.send(message); + } + + for (int i = 0; i < numMessages; i++) + { + for (ClientConsumer consumer : consumers) + { + ClientMessage msg = consumer.receive(RECEIVE_TIMEOUT); + + assertNotNull(msg); + + assertEquals(i, msg.getProperty(new SimpleString("count"))); + + msg.acknowledge(); + } + } + + for (int i = 0; i < numMessages; i++) + { + for (ClientConsumer consumer : consumers) + { + ClientMessage msg = consumer.receiveImmediate(); + + assertNull(msg); + } + } + + sessSend.close(); + for (ClientSession session : sessions) + { + session.close(); + } + + for (int i = 0; i < numSessions; i++) + { + SimpleString subName = new SimpleString("sub" + i); + + s.deleteQueue(subName); + } + + s.close(); + + long end = System.currentTimeMillis(); + + log.info("duration " + (end - start)); + } + + protected void doTestF(final ClientSessionFactory sf) throws Exception + { + long start = System.currentTimeMillis(); + + ClientSession s = sf.createSession(false, false, false); + + final int numMessages = 100; + + final int numSessions = 10; + + Set<ClientConsumer> consumers = new HashSet<ClientConsumer>(); + Set<ClientSession> sessions = new HashSet<ClientSession>(); + + for (int i = 0; i < numSessions; i++) + { + SimpleString subName = new SimpleString("sub" + i); + + ClientSession sessConsume = sf.createSession(false, true, true); + + sessConsume.createQueue(ADDRESS, subName, null, false); + + ClientConsumer consumer = sessConsume.createConsumer(subName); + + consumers.add(consumer); + + sessions.add(sessConsume); + } + + ClientSession sessSend = sf.createSession(false, true, true); + + ClientProducer producer = sessSend.createProducer(ADDRESS); + + for (int i = 0; i < numMessages; i++) + { + ClientMessage message = sessSend.createClientMessage(HornetQTextMessage.TYPE, + false, + 0, + System.currentTimeMillis(), + (byte)1); + message.putIntProperty(new SimpleString("count"), i); + producer.send(message); + } + + for (ClientSession session : sessions) + { + session.start(); + } + + for (int i = 0; i < numMessages; i++) + { + for (ClientConsumer consumer : consumers) + { + ClientMessage msg = consumer.receive(RECEIVE_TIMEOUT); + + if (msg == null) + { + throw new IllegalStateException("Failed to receive message " + i); + } + + assertNotNull(msg); + + assertEquals(i, msg.getProperty(new SimpleString("count"))); + + msg.acknowledge(); + } + } + + for (int i = 0; i < numMessages; i++) + { + for (ClientConsumer consumer : consumers) + { + ClientMessage msg = consumer.receiveImmediate(); + + assertNull(msg); + } + } + + sessSend.close(); + for (ClientSession session : sessions) + { + session.close(); + } + + for (int i = 0; i < numSessions; i++) + { + SimpleString subName = new SimpleString("sub" + i); + + s.deleteQueue(subName); + } + + s.close(); + + assertEquals(1, ((ClientSessionFactoryImpl)sf).numSessions()); + + long end = System.currentTimeMillis(); + + log.info("duration " + (end - start)); + } + + protected void doTestG(final ClientSessionFactory sf) throws Exception + { + long start = System.currentTimeMillis(); + + ClientSession s = sf.createSession(false, false, false); + + final int numMessages = 100; + + final int numSessions = 10; + + Set<ClientConsumer> consumers = new HashSet<ClientConsumer>(); + Set<ClientSession> sessions = new HashSet<ClientSession>(); + + for (int i = 0; i < numSessions; i++) + { + SimpleString subName = new SimpleString("sub" + i); + + ClientSession sessConsume = sf.createSession(false, false, false); + + sessConsume.start(); + + sessConsume.createQueue(ADDRESS, subName, null, false); + + ClientConsumer consumer = sessConsume.createConsumer(subName); + + consumers.add(consumer); + + sessions.add(sessConsume); + } + + ClientSession sessSend = sf.createSession(false, false, false); + + ClientProducer producer = sessSend.createProducer(ADDRESS); + + for (int i = 0; i < numMessages; i++) + { + ClientMessage message = sessSend.createClientMessage(HornetQTextMessage.TYPE, + false, + 0, + System.currentTimeMillis(), + (byte)1); + message.putIntProperty(new SimpleString("count"), i); + producer.send(message); + } + + sessSend.rollback(); + + for (int i = 0; i < numMessages; i++) + { + ClientMessage message = sessSend.createClientMessage(HornetQTextMessage.TYPE, + false, + 0, + System.currentTimeMillis(), + (byte)1); + message.putIntProperty(new SimpleString("count"), i); + producer.send(message); + } + + sessSend.commit(); + + for (int i = 0; i < numMessages; i++) + { + for (ClientConsumer consumer : consumers) + { + ClientMessage msg = consumer.receive(RECEIVE_TIMEOUT); + + assertNotNull(msg); + + assertEquals(i, msg.getProperty(new SimpleString("count"))); + + msg.acknowledge(); + } + } + + for (ClientConsumer consumer : consumers) + { + ClientMessage msg = consumer.receiveImmediate(); + + assertNull(msg); + } + + for (ClientSession session : sessions) + { + session.rollback(); + } + + for (int i = 0; i < numMessages; i++) + { + for (ClientConsumer consumer : consumers) + { + ClientMessage msg = consumer.receive(RECEIVE_TIMEOUT); + + assertNotNull(msg); + + assertEquals(i, msg.getProperty(new SimpleString("count"))); + + msg.acknowledge(); + } + } + + for (int i = 0; i < numMessages; i++) + { + for (ClientConsumer consumer : consumers) + { + ClientMessage msg = consumer.receiveImmediate(); + + assertNull(msg); + } + } + + for (ClientSession session : sessions) + { + session.commit(); + } + + sessSend.close(); + for (ClientSession session : sessions) + { + session.close(); + } + + for (int i = 0; i < numSessions; i++) + { + SimpleString subName = new SimpleString("sub" + i); + + s.deleteQueue(subName); + } + + s.close(); + + long end = System.currentTimeMillis(); + + log.info("duration " + (end - start)); + } + + protected void doTestH(final ClientSessionFactory sf) throws Exception + { + long start = System.currentTimeMillis(); + + ClientSession s = sf.createSession(false, false, false); + + final int numMessages = 100; + + final int numSessions = 10; + + Set<ClientConsumer> consumers = new HashSet<ClientConsumer>(); + Set<ClientSession> sessions = new HashSet<ClientSession>(); + + for (int i = 0; i < numSessions; i++) + { + SimpleString subName = new SimpleString("sub" + i); + + ClientSession sessConsume = sf.createSession(false, false, false); + + sessConsume.createQueue(ADDRESS, subName, null, false); + + ClientConsumer consumer = sessConsume.createConsumer(subName); + + consumers.add(consumer); + + sessions.add(sessConsume); + } + + ClientSession sessSend = sf.createSession(false, false, false); + + ClientProducer producer = sessSend.createProducer(ADDRESS); + + for (int i = 0; i < numMessages; i++) + { + ClientMessage message = sessSend.createClientMessage(HornetQTextMessage.TYPE, + false, + 0, + System.currentTimeMillis(), + (byte)1); + message.putIntProperty(new SimpleString("count"), i); + producer.send(message); + } + + sessSend.rollback(); + + for (int i = 0; i < numMessages; i++) + { + ClientMessage message = sessSend.createClientMessage(HornetQTextMessage.TYPE, + false, + 0, + System.currentTimeMillis(), + (byte)1); + message.putIntProperty(new SimpleString("count"), i); + producer.send(message); + } + + sessSend.commit(); + + for (ClientSession session : sessions) + { + session.start(); + } + + for (int i = 0; i < numMessages; i++) + { + for (ClientConsumer consumer : consumers) + { + ClientMessage msg = consumer.receive(RECEIVE_TIMEOUT); + + assertNotNull(msg); + + assertEquals(i, msg.getProperty(new SimpleString("count"))); + + msg.acknowledge(); + } + } + + for (int i = 0; i < numMessages; i++) + { + for (ClientConsumer consumer : consumers) + { + ClientMessage msg = consumer.receiveImmediate(); + + assertNull(msg); + } + } + + for (ClientSession session : sessions) + { + session.rollback(); + } + + for (int i = 0; i < numMessages; i++) + { + for (ClientConsumer consumer : consumers) + { + ClientMessage msg = consumer.receive(RECEIVE_TIMEOUT); + + assertNotNull(msg); + + assertEquals(i, msg.getProperty(new SimpleString("count"))); + + msg.acknowledge(); + } + } + + for (int i = 0; i < numMessages; i++) + { + for (ClientConsumer consumer : consumers) + { + ClientMessage msg = consumer.receiveImmediate(); + + assertNull(msg); + } + } + + for (ClientSession session : sessions) + { + session.commit(); + } + + sessSend.close(); + for (ClientSession session : sessions) + { + session.close(); + } + + for (int i = 0; i < numSessions; i++) + { + SimpleString subName = new SimpleString("sub" + i); + + s.deleteQueue(subName); + } + + s.close(); + + long end = System.currentTimeMillis(); + + log.info("duration " + (end - start)); + } + + protected void doTestI(final ClientSessionFactory sf) throws Exception + { + ClientSession sessCreate = sf.createSession(false, true, true); + + sessCreate.createQueue(ADDRESS, ADDRESS, null, false); + + ClientSession sess = sf.createSession(false, true, true); + + sess.start(); + + ClientConsumer consumer = sess.createConsumer(ADDRESS); + + ClientProducer producer = sess.createProducer(ADDRESS); + + ClientMessage message = sess.createClientMessage(HornetQTextMessage.TYPE, + false, + 0, + System.currentTimeMillis(), + (byte)1); + producer.send(message); + + ClientMessage message2 = consumer.receive(RECEIVE_TIMEOUT); + + assertNotNull(message2); + + message2.acknowledge(); + + sess.close(); + + sessCreate.deleteQueue(ADDRESS); + + sessCreate.close(); + } + + protected void doTestJ(final ClientSessionFactory sf) throws Exception + { + ClientSession sessCreate = sf.createSession(false, true, true); + + sessCreate.createQueue(ADDRESS, ADDRESS, null, false); + + ClientSession sess = sf.createSession(false, true, true); + + sess.start(); + + ClientConsumer consumer = sess.createConsumer(ADDRESS); + + ClientProducer producer = sess.createProducer(ADDRESS); + + ClientMessage message = sess.createClientMessage(HornetQTextMessage.TYPE, + false, + 0, + System.currentTimeMillis(), + (byte)1); + producer.send(message); + + ClientMessage message2 = consumer.receive(RECEIVE_TIMEOUT); + + assertNotNull(message2); + + message2.acknowledge(); + + sess.close(); + + sessCreate.deleteQueue(ADDRESS); + + sessCreate.close(); + } + + protected void doTestK(final ClientSessionFactory sf) throws Exception + { + ClientSession s = sf.createSession(false, false, false); + + s.createQueue(ADDRESS, ADDRESS, null, false); + + final int numConsumers = 100; + + for (int i = 0; i < numConsumers; i++) + { + ClientConsumer consumer = s.createConsumer(ADDRESS); + + consumer.close(); + } + + s.deleteQueue(ADDRESS); + + s.close(); + } + + protected void doTestL(final ClientSessionFactory sf) throws Exception + { + final int numSessions = 10; + + for (int i = 0; i < numSessions; i++) + { + ClientSession session = sf.createSession(false, false, false); + + session.close(); + } + } + + protected void doTestN(final ClientSessionFactory sf) throws Exception + { + ClientSession sessCreate = sf.createSession(false, true, true); + + sessCreate.createQueue(ADDRESS, new SimpleString(ADDRESS.toString()), null, false); + + ClientSession sess = sf.createSession(false, true, true); + + sess.stop(); + + sess.start(); + + sess.stop(); + + ClientConsumer consumer = sess.createConsumer(new SimpleString(ADDRESS.toString())); + + ClientProducer producer = sess.createProducer(ADDRESS); + + ClientMessage message = sess.createClientMessage(HornetQTextMessage.TYPE, + false, + 0, + System.currentTimeMillis(), + (byte)1); + producer.send(message); + + sess.start(); + + ClientMessage message2 = consumer.receive(RECEIVE_TIMEOUT); + + assertNotNull(message2); + + message2.acknowledge(); + + sess.stop(); + + sess.start(); + + sess.close(); + + sessCreate.deleteQueue(new SimpleString(ADDRESS.toString())); + + sessCreate.close(); + } + + protected int getNumIterations() + { + return 2; + } + + protected void setUp() throws Exception + { + super.setUp(); + + timer = new Timer(true); + } + + protected void tearDown() throws Exception + { + timer.cancel(); + + InVMRegistry.instance.clear(); + + super.tearDown(); + } + + // Private ------------------------------------------------------- + + private Failer startFailer(final long time, final ClientSession session) + { + Failer failer = new Failer((ClientSessionInternal)session); + + timer.schedule(failer, (long)(time * Math.random()), 100); + + return failer; + } + + private void start() throws Exception + { + Configuration liveConf = new ConfigurationImpl(); + liveConf.setSecurityEnabled(false); + liveConf.getAcceptorConfigurations() + .add(new TransportConfiguration("org.hornetq.core.remoting.impl.invm.InVMAcceptorFactory")); + liveService = HornetQ.newHornetQServer(liveConf, false); + liveService.start(); + } + + private void stop() throws Exception + { + liveService.stop(); + + assertEquals(0, InVMRegistry.instance.size()); + + liveService = null; + } + + // Inner classes ------------------------------------------------- + + class Failer extends TimerTask + { + private final ClientSessionInternal session; + + private boolean executed; + + public Failer(final ClientSessionInternal session) + { + this.session = session; + } + + public synchronized void run() + { + log.info("** Failing connection"); + + session.getConnection().fail(new HornetQException(HornetQException.NOT_CONNECTED, "oops")); + + log.info("** Fail complete"); + + cancel(); + + executed = true; + } + + public synchronized boolean isExecuted() + { + return executed; + } + } + + public abstract class RunnableT + { + abstract void run(final ClientSessionFactory sf) throws Exception; + } +} Copied: branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/integration/cluster/reattach/ReattachTest.java (from rev 7962, branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/integration/cluster/failover/ReattachTest.java) =================================================================== --- branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/integration/cluster/reattach/ReattachTest.java (rev 0) +++ branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/integration/cluster/reattach/ReattachTest.java 2009-09-22 15:08:41 UTC (rev 7979) @@ -0,0 +1,722 @@ +/* + * Copyright 2009 Red Hat, Inc. + * Red Hat licenses this file to you under the Apache License, version + * 2.0 (the "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * http://www.apache.org/licenses/LICENSE-2.0 + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or + * implied. See the License for the specific language governing + * permissions and limitations under the License. + */ + +package org.hornetq.tests.integration.cluster.reattach; + +import org.hornetq.core.client.ClientConsumer; +import org.hornetq.core.client.ClientMessage; +import org.hornetq.core.client.ClientProducer; +import org.hornetq.core.client.ClientSession; +import org.hornetq.core.client.impl.ClientSessionFactoryImpl; +import org.hornetq.core.client.impl.ClientSessionFactoryInternal; +import org.hornetq.core.client.impl.ClientSessionInternal; +import org.hornetq.core.config.Configuration; +import org.hornetq.core.config.TransportConfiguration; +import org.hornetq.core.config.impl.ConfigurationImpl; +import org.hornetq.core.exception.HornetQException; +import org.hornetq.core.logging.Logger; +import org.hornetq.core.remoting.FailureListener; +import org.hornetq.core.remoting.RemotingConnection; +import org.hornetq.core.remoting.impl.invm.InVMConnector; +import org.hornetq.core.remoting.impl.invm.InVMRegistry; +import org.hornetq.core.server.HornetQ; +import org.hornetq.core.server.HornetQServer; +import org.hornetq.jms.client.HornetQTextMessage; +import org.hornetq.tests.util.UnitTestCase; +import org.hornetq.utils.SimpleString; + +/** + * + * A ReattachTest + * + * @author <a href="mailto:tim.fox@jboss.com">Tim Fox</a> + * + * Created 4 Nov 2008 16:54:50 + * + * + */ +public class ReattachTest extends UnitTestCase +{ + private static final Logger log = Logger.getLogger(ReattachTest.class); + + // Constants ----------------------------------------------------- + + // Attributes ---------------------------------------------------- + + private static final SimpleString ADDRESS = new SimpleString("FailoverTestAddress"); + + private HornetQServer service; + + // Static -------------------------------------------------------- + + // Constructors -------------------------------------------------- + + // Public -------------------------------------------------------- + + /* + * Test failure on connection, but server is still up so should immediately reconnect + */ + public void testImmediateReattach() throws Exception + { + final long retryInterval = 500; + + final double retryMultiplier = 1d; + + final int reconnectAttempts = 1; + + ClientSessionFactoryInternal sf = new ClientSessionFactoryImpl(new TransportConfiguration("org.hornetq.core.remoting.impl.invm.InVMConnectorFactory")); + + sf.setRetryInterval(retryInterval); + sf.setRetryIntervalMultiplier(retryMultiplier); + sf.setReconnectAttempts(reconnectAttempts); + sf.setUseReattach(true); + + ClientSession session = sf.createSession(false, true, true); + + session.createQueue(ADDRESS, ADDRESS, null, false); + + final int numIterations = 100; + + for (int j = 0; j < numIterations; j++) + { + ClientProducer producer = session.createProducer(ADDRESS); + + final int numMessages = 1000; + + for (int i = 0; i < numMessages; i++) + { + ClientMessage message = session.createClientMessage(HornetQTextMessage.TYPE, + false, + 0, + System.currentTimeMillis(), + (byte)1); + message.putIntProperty(new SimpleString("count"), i); + message.getBody().writeString("aardvarks"); + producer.send(message); + } + + ClientConsumer consumer = session.createConsumer(ADDRESS); + + RemotingConnection conn = ((ClientSessionInternal)session).getConnection(); + + conn.fail(new HornetQException(HornetQException.NOT_CONNECTED)); + + session.start(); + + for (int i = 0; i < numMessages; i++) + { + ClientMessage message = consumer.receive(500); + + assertNotNull(message); + + assertEquals("aardvarks", message.getBody().readString()); + + assertEquals(i, message.getProperty(new SimpleString("count"))); + + message.acknowledge(); + } + + ClientMessage message = consumer.receiveImmediate(); + + assertNull(message); + + producer.close(); + + consumer.close(); + } + + session.close(); + + sf.close(); + } + + /* + * Test failure on connection, simulate failure to create connection for a while, then + * allow connection to be recreated + */ + public void testDelayedReattach() throws Exception + { + final long retryInterval = 500; + + final double retryMultiplier = 1d; + + final int reconnectAttempts = -1; + + ClientSessionFactoryInternal sf = new ClientSessionFactoryImpl(new TransportConfiguration("org.hornetq.core.remoting.impl.invm.InVMConnectorFactory")); + + sf.setRetryInterval(retryInterval); + sf.setRetryIntervalMultiplier(retryMultiplier); + sf.setReconnectAttempts(reconnectAttempts); + sf.setUseReattach(true); + + ClientSession session = sf.createSession(false, true, true); + + session.createQueue(ADDRESS, ADDRESS, null, false); + + ClientProducer producer = session.createProducer(ADDRESS); + + final int numMessages = 1000; + + for (int i = 0; i < numMessages; i++) + { + ClientMessage message = session.createClientMessage(HornetQTextMessage.TYPE, + false, + 0, + System.currentTimeMillis(), + (byte)1); + message.putIntProperty(new SimpleString("count"), i); + message.getBody().writeString("aardvarks"); + producer.send(message); + } + + ClientConsumer consumer = session.createConsumer(ADDRESS); + + InVMConnector.failOnCreateConnection = true; + + RemotingConnection conn = ((ClientSessionInternal)session).getConnection(); + + Thread t = new Thread() + { + public void run() + { + try + { + Thread.sleep(retryInterval * 3); + } + catch (InterruptedException ignore) + { + } + + InVMConnector.failOnCreateConnection = false; + } + }; + + t.start(); + + conn.fail(new HornetQException(HornetQException.NOT_CONNECTED)); + + session.start(); + + for (int i = 0; i < numMessages; i++) + { + ClientMessage message = consumer.receive(500); + + assertNotNull(message); + + assertEquals("aardvarks", message.getBody().readString()); + + assertEquals(i, message.getProperty(new SimpleString("count"))); + + message.acknowledge(); + } + + ClientMessage message = consumer.receiveImmediate(); + + assertNull(message); + + session.close(); + + sf.close(); + + t.join(); + } + + // Test an async (e.g. pinger) failure coming in while a connection manager is already reconnecting + public void testAsyncFailureWhileReattaching() throws Exception + { + final long retryInterval = 500; + + final double retryMultiplier = 1d; + + final int reconnectAttempts = -1; + + final long asyncFailDelay = 2000; + + ClientSessionFactoryInternal sf = new ClientSessionFactoryImpl(new TransportConfiguration("org.hornetq.core.remoting.impl.invm.InVMConnectorFactory")); + + sf.setRetryInterval(retryInterval); + sf.setRetryIntervalMultiplier(retryMultiplier); + sf.setReconnectAttempts(reconnectAttempts); + sf.setUseReattach(true); + + ClientSession session = sf.createSession(false, true, true); + + ClientSession session2 = sf.createSession(false, true, true); + + class MyFailureListener implements FailureListener + { + volatile boolean failed; + + public void connectionFailed(HornetQException me) + { + failed = true; + } + } + + MyFailureListener listener = new MyFailureListener(); + + session2.addFailureListener(listener); + + session.createQueue(ADDRESS, ADDRESS, null, false); + + ClientProducer producer = session.createProducer(ADDRESS); + + final int numMessages = 1000; + + for (int i = 0; i < numMessages; i++) + { + ClientMessage message = session.createClientMessage(HornetQTextMessage.TYPE, + false, + 0, + System.currentTimeMillis(), + (byte)1); + message.putIntProperty(new SimpleString("count"), i); + message.getBody().writeString("aardvarks"); + producer.send(message); + } + + ClientConsumer consumer = session.createConsumer(ADDRESS); + + InVMConnector.numberOfFailures = 10; + InVMConnector.failOnCreateConnection = true; + + //We need to fail on different connections. + + //We fail on one connection then the connection manager tries to reconnect all connections + //Then we fail the other, and the connection manager is then called while the reconnection is occurring + //We can't use the same connection since RemotingConnectionImpl only allows one fail to be in process + //at same time + + final RemotingConnection conn = ((ClientSessionInternal)session).getConnection(); + + final RemotingConnection conn2 = ((ClientSessionInternal)session2).getConnection(); + + assertTrue(conn != conn2); + + Thread t = new Thread() + { + public void run() + { + try + { + Thread.sleep(asyncFailDelay); + } + catch (InterruptedException ignore) + { + } + + log.info("calling fail async"); + + conn2.fail(new HornetQException(HornetQException.NOT_CONNECTED, "Did not receive pong from server")); + } + }; + + t.start(); + + conn.fail(new HornetQException(HornetQException.NOT_CONNECTED)); + + assertTrue(listener.failed); + + session.start(); + + for (int i = 0; i < numMessages; i++) + { + ClientMessage message = consumer.receive(500); + + assertNotNull(message); + + assertEquals("aardvarks", message.getBody().readString()); + + assertEquals(i, message.getProperty(new SimpleString("count"))); + + message.acknowledge(); + } + + ClientMessage message = consumer.receiveImmediate(); + + assertNull(message); + + session.close(); + + session2.close(); + + sf.close(); + + t.join(); + } + + public void testReattachAttemptsFailsToReconnect() throws Exception + { + final long retryInterval = 500; + + final double retryMultiplier = 1d; + + final int reconnectAttempts = 3; + + ClientSessionFactoryInternal sf = new ClientSessionFactoryImpl(new TransportConfiguration("org.hornetq.core.remoting.impl.invm.InVMConnectorFactory")); + + sf.setRetryInterval(retryInterval); + sf.setRetryIntervalMultiplier(retryMultiplier); + sf.setReconnectAttempts(reconnectAttempts); + sf.setUseReattach(true); + + ClientSession session = sf.createSession(false, true, true); + + session.createQueue(ADDRESS, ADDRESS, null, false); + + ClientProducer producer = session.createProducer(ADDRESS); + + final int numMessages = 1000; + + for (int i = 0; i < numMessages; i++) + { + ClientMessage message = session.createClientMessage(HornetQTextMessage.TYPE, + false, + 0, + System.currentTimeMillis(), + (byte)1); + message.putIntProperty(new SimpleString("count"), i); + message.getBody().writeString("aardvarks"); + producer.send(message); + } + + ClientConsumer consumer = session.createConsumer(ADDRESS); + + InVMConnector.failOnCreateConnection = true; + + RemotingConnection conn = ((ClientSessionInternal)session).getConnection(); + + // Sleep for longer than max retries so should fail to reconnect + + Thread t = new Thread() + { + public void run() + { + try + { + Thread.sleep(retryInterval * (reconnectAttempts + 1)); + } + catch (InterruptedException ignore) + { + } + + InVMConnector.failOnCreateConnection = false; + } + }; + + t.start(); + + conn.fail(new HornetQException(HornetQException.NOT_CONNECTED)); + + session.start(); + + // Should be null since failed to reconnect + ClientMessage message = consumer.receive(500); + + assertNull(message); + + session.close(); + + sf.close(); + + t.join(); + } + + public void testReattachAttemptsSucceedsInReconnecting() throws Exception + { + final long retryInterval = 500; + + final double retryMultiplier = 1d; + + final int reconnectAttempts = 10; + + ClientSessionFactoryInternal sf = new ClientSessionFactoryImpl(new TransportConfiguration("org.hornetq.core.remoting.impl.invm.InVMConnectorFactory")); + + sf.setRetryInterval(retryInterval); + sf.setRetryIntervalMultiplier(retryMultiplier); + sf.setReconnectAttempts(reconnectAttempts); + sf.setUseReattach(true); + + ClientSession session = sf.createSession(false, true, true); + + session.createQueue(ADDRESS, ADDRESS, null, false); + + ClientProducer producer = session.createProducer(ADDRESS); + + final int numMessages = 1000; + + for (int i = 0; i < numMessages; i++) + { + ClientMessage message = session.createClientMessage(HornetQTextMessage.TYPE, + false, + 0, + System.currentTimeMillis(), + (byte)1); + message.putIntProperty(new SimpleString("count"), i); + message.getBody().writeString("aardvarks"); + producer.send(message); + } + + ClientConsumer consumer = session.createConsumer(ADDRESS); + + InVMConnector.failOnCreateConnection = true; + InVMConnector.numberOfFailures = reconnectAttempts - 1; + + RemotingConnection conn = ((ClientSessionInternal)session).getConnection(); + + conn.fail(new HornetQException(HornetQException.NOT_CONNECTED)); + + session.start(); + + for (int i = 0; i < numMessages; i++) + { + ClientMessage message = consumer.receive(500); + + assertNotNull(message); + + assertEquals("aardvarks", message.getBody().readString()); + + assertEquals(i, message.getProperty(new SimpleString("count"))); + + message.acknowledge(); + } + + ClientMessage message = consumer.receiveImmediate(); + + assertNull(message); + + session.close(); + + sf.close(); + } + + public void testRetryInterval() throws Exception + { + final long retryInterval = 500; + + final double retryMultiplier = 1d; + + final int reconnectAttempts = -1; + + ClientSessionFactoryInternal sf = new ClientSessionFactoryImpl(new TransportConfiguration("org.hornetq.core.remoting.impl.invm.InVMConnectorFactory")); + + sf.setRetryInterval(retryInterval); + sf.setRetryIntervalMultiplier(retryMultiplier); + sf.setReconnectAttempts(reconnectAttempts); + sf.setUseReattach(true); + + ClientSession session = sf.createSession(false, true, true); + + session.createQueue(ADDRESS, ADDRESS, null, false); + + ClientProducer producer = session.createProducer(ADDRESS); + + final int numMessages = 1000; + + for (int i = 0; i < numMessages; i++) + { + ClientMessage message = session.createClientMessage(HornetQTextMessage.TYPE, + false, + 0, + System.currentTimeMillis(), + (byte)1); + message.putIntProperty(new SimpleString("count"), i); + message.getBody().writeString("aardvarks"); + producer.send(message); + } + + ClientConsumer consumer = session.createConsumer(ADDRESS); + + InVMConnector.failOnCreateConnection = true; + + RemotingConnection conn = ((ClientSessionInternal)session).getConnection(); + + long start = System.currentTimeMillis(); + + Thread t = new Thread() + { + public void run() + { + try + { + Thread.sleep(retryInterval / 2); + } + catch (InterruptedException ignore) + { + } + InVMConnector.failOnCreateConnection = false; + } + }; + + t.start(); + + conn.fail(new HornetQException(HornetQException.NOT_CONNECTED)); + + session.start(); + + for (int i = 0; i < numMessages; i++) + { + ClientMessage message = consumer.receive(500); + + assertNotNull(message); + + assertEquals("aardvarks", message.getBody().readString()); + + assertEquals(i, message.getProperty(new SimpleString("count"))); + + message.acknowledge(); + } + + ClientMessage message = consumer.receiveImmediate(); + + assertNull(message); + + long end = System.currentTimeMillis(); + + assertTrue((end - start) >= retryInterval); + + session.close(); + + sf.close(); + + t.join(); + } + + public void testExponentialBackoff() throws Exception + { + final long retryInterval = 500; + + final double retryMultiplier = 4d; + + final int reconnectAttempts = -1; + + ClientSessionFactoryInternal sf = new ClientSessionFactoryImpl(new TransportConfiguration("org.hornetq.core.remoting.impl.invm.InVMConnectorFactory")); + + sf.setRetryInterval(retryInterval); + sf.setRetryIntervalMultiplier(retryMultiplier); + sf.setReconnectAttempts(reconnectAttempts); + sf.setUseReattach(true); + + ClientSession session = sf.createSession(false, true, true); + + session.createQueue(ADDRESS, ADDRESS, null, false); + + ClientProducer producer = session.createProducer(ADDRESS); + + final int numMessages = 1000; + + for (int i = 0; i < numMessages; i++) + { + ClientMessage message = session.createClientMessage(HornetQTextMessage.TYPE, + false, + 0, + System.currentTimeMillis(), + (byte)1); + message.putIntProperty(new SimpleString("count"), i); + message.getBody().writeString("aardvarks"); + producer.send(message); + } + + ClientConsumer consumer = session.createConsumer(ADDRESS); + + InVMConnector.failOnCreateConnection = true; + + RemotingConnection conn = ((ClientSessionInternal)session).getConnection(); + + long start = System.currentTimeMillis(); + + Thread t = new Thread() + { + public void run() + { + try + { + Thread.sleep(retryInterval * 2); + } + catch (InterruptedException ignore) + { + } + + InVMConnector.failOnCreateConnection = false; + } + }; + + t.start(); + + conn.fail(new HornetQException(HornetQException.NOT_CONNECTED)); + + session.start(); + + for (int i = 0; i < numMessages; i++) + { + ClientMessage message = consumer.receive(500); + + assertNotNull(message); + + assertEquals("aardvarks", message.getBody().readString()); + + assertEquals(i, message.getProperty(new SimpleString("count"))); + + message.acknowledge(); + } + + ClientMessage message = consumer.receiveImmediate(); + + assertNull(message); + + long end = System.currentTimeMillis(); + + assertTrue((end - start) >= retryInterval * (1 + retryMultiplier)); + + session.close(); + + sf.close(); + + t.join(); + } + + // Package protected --------------------------------------------- + + // Protected ----------------------------------------------------- + + @Override + protected void setUp() throws Exception + { + super.setUp(); + + Configuration liveConf = new ConfigurationImpl(); + liveConf.setSecurityEnabled(false); + liveConf.getAcceptorConfigurations() + .add(new TransportConfiguration("org.hornetq.core.remoting.impl.invm.InVMAcceptorFactory")); + service = HornetQ.newHornetQServer(liveConf, false); + service.start(); + } + + @Override + protected void tearDown() throws Exception + { + InVMConnector.resetFailures(); + + service.stop(); + + assertEquals(0, InVMRegistry.instance.size()); + + service = null; + + super.tearDown(); + } + + // Private ------------------------------------------------------- + + // Inner classes ------------------------------------------------- +} Modified: branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/soak/failover/RandomFailoverSoakTest.java =================================================================== --- branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/soak/failover/RandomFailoverSoakTest.java 2009-09-22 13:57:55 UTC (rev 7978) +++ branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/soak/failover/RandomFailoverSoakTest.java 2009-09-22 15:08:41 UTC (rev 7979) @@ -14,7 +14,7 @@ package org.hornetq.tests.soak.failover; -import org.hornetq.tests.integration.cluster.failover.RandomReattachTest; +import org.hornetq.tests.integration.cluster.reattach.RandomReattachTest; /** * A RandomFailoverSoakTest Modified: branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/stress/failover/MultiThreadRandomFailoverStressTest.java =================================================================== --- branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/stress/failover/MultiThreadRandomFailoverStressTest.java 2009-09-22 13:57:55 UTC (rev 7978) +++ branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/stress/failover/MultiThreadRandomFailoverStressTest.java 2009-09-22 15:08:41 UTC (rev 7979) @@ -13,7 +13,7 @@ package org.hornetq.tests.stress.failover; -import org.hornetq.tests.integration.cluster.failover.MultiThreadRandomReattachTest; +import org.hornetq.tests.integration.cluster.reattach.MultiThreadRandomReattachTest; /** * A MultiThreadRandomFailoverStressTest Modified: branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/stress/failover/RandomFailoverStressTest.java =================================================================== --- branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/stress/failover/RandomFailoverStressTest.java 2009-09-22 13:57:55 UTC (rev 7978) +++ branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/stress/failover/RandomFailoverStressTest.java 2009-09-22 15:08:41 UTC (rev 7979) @@ -14,7 +14,7 @@ package org.hornetq.tests.stress.failover; -import org.hornetq.tests.integration.cluster.failover.RandomReattachTest; +import org.hornetq.tests.integration.cluster.reattach.RandomReattachTest; /** * A RandomFailoverStressTest