foundation/eclipselink.core.test/src/it/java/org/eclipse/persistence/testing/tests/distributedservers/rcm/broadcast/BroadcastReconnectionTest.java - eclipselink - Git at Google

 /*
  * Copyright (c) 1998, 2021 Oracle and/or its affiliates. All rights reserved.
  *
  * This program and the accompanying materials are made available under the
  * terms of the Eclipse Public License v. 2.0 which is available at
  * http://www.eclipse.org/legal/epl-2.0,
  * or the Eclipse Distribution License v. 1.0 which is available at
  * http://www.eclipse.org/org/documents/edl-v10.php.
  *
  * SPDX-License-Identifier: EPL-2.0 OR BSD-3-Clause
  */

 // Contributors:
 //     Oracle - initial API and implementation from Oracle TopLink
 package org.eclipse.persistence.testing.tests.distributedservers.rcm.broadcast;

 import java.util.ArrayList;
 import java.util.Iterator;

 import org.eclipse.persistence.exceptions.ExceptionHandler;
 import org.eclipse.persistence.exceptions.RemoteCommandManagerException;
 import org.eclipse.persistence.internal.sessions.AbstractSession;
 import org.eclipse.persistence.sessions.coordination.TransportManager;
 import org.eclipse.persistence.testing.framework.TestErrorException;
 import org.eclipse.persistence.testing.framework.TestWrapper;

 public class BroadcastReconnectionTest extends TestWrapper {
     BroadcastSetupHelper helper;
     BroadcastEventLock eventLock;
     ArrayList localConnectionCreators;

     boolean shouldDestroyFactory;
     boolean shouldRemoveConnectionOnError;
     boolean shouldRemoveConnectionOnErrorOriginal;

     boolean sourceHasThrownErrorPropagatingCommandException;
     boolean sourceHasRemovedRemoteConnection;

     Exception firstRunTestException;
     Exception firstRunVerifyException;

     // take any test that tests sending and receiving remote command

     public BroadcastReconnectionTest(BroadcastSetupHelper.TestWrapperWithEventLock test, boolean shouldDestroyFactory, boolean shouldRemoveConnectionOnError, BroadcastSetupHelper helper) {
         super(test);
         this.helper = helper;
         this.eventLock = test.getEventLock();

         this.shouldDestroyFactory = shouldDestroyFactory;
         this.shouldRemoveConnectionOnError = shouldRemoveConnectionOnError;
         setName("BroadcastReconnectionTest: shouldDestroyFactory = " + shouldDestroyFactory + "; shouldRemoveConnectionOnError = " + shouldRemoveConnectionOnError);
         setDescription("Invalidates RemoteConnection then removes it and runs cache sync test again");
     }

     protected TransportManager getTransportManager() {
         return ((AbstractSession)getSession()).getCommandManager().getTransportManager();
     }

     // Exception handler tries to re-creates local connection in a new thread.
     // Used only in case shouldRemoveConnectionOnError==true and
     // there is an exception thrown by local connection while it listens:
     // that happens in JMS case, doesn't happen with Oc4jJGroups.

     class LocalConnectionCreator implements ExceptionHandler, Runnable {
         // implements ExceptionHandler - used only in case shouldRemoveConnectionOnError==true
         AbstractSession session;
         ExceptionHandler originalExceptionHandler;
         boolean isActive;
         boolean hasReconnected;

         LocalConnectionCreator(AbstractSession session) {
             this.originalExceptionHandler = session.getExceptionHandler();
             session.setExceptionHandler(this);
             this.session = session;
         }

         @Override
         public Object handleException(RuntimeException exception) {
             if (exception instanceof RemoteCommandManagerException) {
                 if (((RemoteCommandManagerException)exception).getErrorCode() == helper.getRcmExceptionErrorCodeOnFailureToCreateLocalConnection()) {
                     if (isActive) {
                         // already trying to reconnect - eat the exception.
                     } else {
                         isActive = true;
                         hasReconnected = false;
                         session.getServerPlatform().launchContainerRunnable(this);
                     }
                     return null;
                 }
             }
             if (originalExceptionHandler != null) {
                 return originalExceptionHandler.handleException(exception);
             } else {
                 throw exception;
             }
         }

         @Override
         public void run() {
             while (isActive) {
                 // no need for try block - we are eating the exception (see handleException method)
                 session.getCommandManager().getTransportManager().createLocalConnection();
                 if (session.getCommandManager().getTransportManager().getConnectionToLocalHost() != null) {
                     // success!
                     isActive = false;
                     hasReconnected = true;
                 }
             }
         }

         void clear() {
             if (isActive) {
                 isActive = false;
             }
             session.setExceptionHandler(originalExceptionHandler);
         }
     }

     @Override
     protected void setup() throws Throwable {
         // save originals to restore back in reset
         shouldRemoveConnectionOnErrorOriginal = getTransportManager().shouldRemoveConnectionOnError();
         // set the new values for duration of the test
         getTransportManager().setShouldRemoveConnectionOnError(shouldRemoveConnectionOnError);
         // JMS only: if creation of local (listening) JMS connection fails
         // (which it will after the factory is destroyed) then keep trying to re-create listening connection.
         // Should succeed after the factory is recreated.
         // Note that there's no need for that for the main session (the one that sends messages):
         // the listening connection for it will be recreated together with the sending connection -
         // on attempt to send a message.
         if (shouldRemoveConnectionOnError && helper.isLocalConnectionRemovedOnListeningError()) {
             localConnectionCreators = new ArrayList();
             Iterator it = helper.getSessionsIterator();
             while (it.hasNext()) {
                 AbstractSession session = (AbstractSession)it.next();
                 if (session != getSession()) {
                     localConnectionCreators.add(new LocalConnectionCreator(session));
                 }
             }
         }
         // Shutdown the factory - after this sending and receiving remote commands should fail
         if (shouldDestroyFactory) {
             helper.destroyFactory();
         } else {
             stopFactory();
         }
         // disable the target listener.
         // need this in Oc4jJGroups case: the message sent through closed (or destroyed)
         // factory is still delivered to the target - and it happens before the
         // exception is thrown on the source.
         // EventLock would only accept the first unlocking event - and therefore unless
         // UNLOCKED_BY_TARGET_LISTENER is disabled exception (or connection removal)
         // would not show up as a EventLock's state after the test is complete.
         if (helper.shouldIgnoreTargetListenerInReconnectionTest()) {
             eventLock.disableState(BroadcastEventLock.UNLOCKED_BY_TARGET_LISTENER);
         }

         // setup the internal test for the first run
         super.setup();
     }

     @Override
     protected void test() throws Throwable {
         try {
             // This is expected to fail - the factory has been shut down.
             super.test();
         } catch (Exception ex) {
             // Because message published in a separate thread no exception is thrown in the main thread.
             // Exception is never thrown
             firstRunTestException = ex;
         }
         // Because the factory was either stopped or destroyed verify should fail.
         // However eventLock allows the wrapped test to proceed as soon as either
         // exception is thrown or connection is removed on the source - therefore there's no guarantee
         // that the message failed to be sent by the source will fail to be eventually recieved by the target -
         // may be if we waited a little bit more it would have reached the target.
         // It's not the case with OracleAQ-based JMS, but it seem to happen with Oc4jJGroups:
         // sending message throws exception on the source, but the message still reaches the target.
         try {
             super.verify();
         } catch (Exception ex) {
             // ignore
             firstRunVerifyException = ex;
         } finally {
             // reset the internal test - the first run is complete
             super.reset();
         }

         // look at the state of eventLock to see whether any events of interest have occurred during super.test()
         int state = eventLock.getState();

         // should be equal to !shouldRemoveConnectionOnError:
         // exception is thrown only in case connection is NOT to be removed on error
         sourceHasThrownErrorPropagatingCommandException = state == BroadcastEventLock.UNLOCKED_BY_SOURCE_EXCEPTION_HANDLER;
         // should be equal to shouldRemoveConnectionOnError:
         // connection is removed only in case connection is to be removed on error
         sourceHasRemovedRemoteConnection = state == BroadcastEventLock.UNLOCKED_BY_SOURCE_SESSION;

         // repair connections and recreate the factory in case the factory was destroyed
         if (shouldDestroyFactory) {
             // repair all connections, restart factory.
             resetConnections();
         } else {
             // restart the factory
             helper.startFactory();
         }

         if (shouldRemoveConnectionOnError || shouldDestroyFactory) {
             // In External connection has been removed either as a result of the first run,
             // or by restConnection method,
             // therefore the second run will be immediately unlocked with
             // UNLOCKED_BY_SOURCE_SESSION unless we disable it.
             eventLock.disableState(BroadcastEventLock.UNLOCKED_BY_SOURCE_SESSION);
         }

         // now attempt to run the internal test again -
         // connections have been repaired therefore now it should pass.
         super.setup();
         super.test();
     }

     @Override
     protected void verify() throws Throwable {
         if (shouldRemoveConnectionOnError) {
             if (sourceHasThrownErrorPropagatingCommandException) {
                 throw new TestErrorException("With shouldRemoveConnectionOnError==true there should've been NO ErrorPropagatingCommand RCMException thrown");
             }
             if (!sourceHasRemovedRemoteConnection) {
                 throw new TestErrorException("With shouldRemoveConnectionOnError==true remote connection should've been removed");
             }
         } else {
             if (!sourceHasThrownErrorPropagatingCommandException) {
                 throw new TestErrorException("With shouldRemoveConnectionOnError==false there should've been ErrorPropagatingCommand RCMException thrown");
             }
             if (sourceHasRemovedRemoteConnection) {
                 throw new TestErrorException("With shouldRemoveConnectionOnError==false remote connection should've NOT been removed");
             }
         }
         if (firstRunTestException != null) {
             throw new TestErrorException("Unexpectedly firstRunTestException was thrown: ", firstRunTestException);
         }

         // look at the state of eventLock to see whether any events of interest have occurred during super.test()
         int state = eventLock.getState();
         if (helper.shouldIgnoreTargetListenerInReconnectionTest()) {
             // because targetListener is disabled, the successful result would unlock by timer -
             // that means there were no exceptions or connection removals.
             if (state != BroadcastEventLock.UNLOCKED_BY_TIMER) {
                 throw new TestErrorException("Unexpected state " + state + " after the second run, BroadcastEventLock.UNLOCKED_BY_TIMER was expected");
             }
         } else {
             if (state != BroadcastEventLock.UNLOCKED_BY_TARGET_LISTENER) {
                 throw new TestErrorException("Unexpected state " + state + " after the second run, BroadcastEventLock.UNLOCKED_BY_TARGET_LISTENER was expected");
             }
             if (firstRunVerifyException == null) {
                 throw new TestErrorException("Unexpectedly firstRunVerifyException was NOT thrown");
             }
         }

         // verify the second run of internal test - now it should pass
         super.verify();
     }

     @Override
     public void reset() throws Throwable {
         if (localConnectionCreators != null) {
             Iterator it = localConnectionCreators.iterator();
             while (it.hasNext()) {
                 LocalConnectionCreator localConnectionCreator = (LocalConnectionCreator)it.next();
                 localConnectionCreator.clear();
             }
             localConnectionCreators = null;
         }
         if (shouldRemoveConnectionOnError) {
             // External connections for sessions other than main are still invalid - let's remove them.
             // Otherwise in case shouldRemoveConnectionOnError is set to false (either here or in another test)
             // the sessions will be stuck with the dead connections.
             helper.removeConnectionsForAllSessionsExcept((AbstractSession)getSession(), "external");
         }
         getTransportManager().setShouldRemoveConnectionOnError(shouldRemoveConnectionOnErrorOriginal);
         firstRunTestException = null;
         firstRunVerifyException = null;
         // some states could have been disabled - set back the original states' enabling.
         eventLock.enableAllStates();
         // reset internal test
         super.reset();
     }

     protected void stopFactory() throws Exception {
         helper.stopFactory();
     }

     protected void resetConnections() throws Exception {
         // the "main" session casted to AbstractSession
         AbstractSession abstractSession = (AbstractSession)getSession();

         // remove dead connections that haven't been already automatically removed.
         if (shouldRemoveConnectionOnError) {
             // getSession()'s external connection is already removed.
             // It will be automatically repaired next time it attempts sending a message.

             // External connections for all other sessions are still there:
             // they didn't send messages => haven't thrown an error => haven't been removed.
             // In these tests other sessions never send any messages, therefore we
             // don't really care about their external connections. However would the
             // other sessions attempt to send a message their invalid external connection
             // would throw exception and will be substituted with the new connection
             // (provided shouldRemoveConnectionOnError is still set to true).

             // remove local connections.
             if (helper.isLocalConnectionRemovedOnListeningError()) {
                 // JMS case.
                 // All local connections should be removed on listening error -
                 // subscriber.receive method throwing JMSException. Nothing to do.
             } else {
                 if (helper.isLocalConnectionAlsoExternalConnection()) {
                     // Oc4jJGroups case - local connections for sessions other than the main one
                     // should be removed.
                     // The local connection is the same as external connection:
                     // external connection for the main session was removed => local was removed, too.
                     helper.removeConnectionsForAllSessionsExcept(abstractSession, "local");
                 } else {
                     // Currently never get here - provided for consistency only.
                     helper.removeConnectionsForAllSessions("local");
                 }
             }
         } else {
             // remove all connections for all sessions.
             helper.removeConnectionsForAllSessions("all");
         }
         // recreate the factory
         helper.createFactory();
         // start the factory
         helper.startFactory();

         // Re-create connections except those that will be re-created automatically
         // (those could be re-created "by hand", too - but that's not necessary).

         // No need to recreate external connections - they are automatically
         // re-creted on attempt to send a message.

         // Recreate local connections.
         if (localConnectionCreators != null) {
             // JMS shouldRemoveConnectionOnError==true case:
             // local connections on sessions other than main should be restored by exception handlers -
             // just wait until they are done.
             boolean allReconnected;
             do {
                 allReconnected = true;
                 Iterator it = localConnectionCreators.iterator();
                 while (it.hasNext()) {
                     LocalConnectionCreator localConnectionCreator = (LocalConnectionCreator)it.next();
                     allReconnected = allReconnected && localConnectionCreator.hasReconnected;
                 }
                 if (!allReconnected) {
                     Thread.sleep(1000);
                 }
             } while (!allReconnected);
         } else {
             // for main session localConnection will be created simultaneously with the external one:
             // on attempt to send a message.
             helper.createConnectionsForAllSessionsExcept(abstractSession, "local");
         }
     }
 }
	/*
	* Copyright (c) 1998, 2021 Oracle and/or its affiliates. All rights reserved.
	*
	* This program and the accompanying materials are made available under the
	* terms of the Eclipse Public License v. 2.0 which is available at
	* http://www.eclipse.org/legal/epl-2.0,
	* or the Eclipse Distribution License v. 1.0 which is available at
	* http://www.eclipse.org/org/documents/edl-v10.php.
	*
	* SPDX-License-Identifier: EPL-2.0 OR BSD-3-Clause
	*/

	// Contributors:
	// Oracle - initial API and implementation from Oracle TopLink
	package org.eclipse.persistence.testing.tests.distributedservers.rcm.broadcast;

	import java.util.ArrayList;
	import java.util.Iterator;

	import org.eclipse.persistence.exceptions.ExceptionHandler;
	import org.eclipse.persistence.exceptions.RemoteCommandManagerException;
	import org.eclipse.persistence.internal.sessions.AbstractSession;
	import org.eclipse.persistence.sessions.coordination.TransportManager;
	import org.eclipse.persistence.testing.framework.TestErrorException;
	import org.eclipse.persistence.testing.framework.TestWrapper;

	public class BroadcastReconnectionTest extends TestWrapper {
	BroadcastSetupHelper helper;
	BroadcastEventLock eventLock;
	ArrayList localConnectionCreators;

	boolean shouldDestroyFactory;
	boolean shouldRemoveConnectionOnError;
	boolean shouldRemoveConnectionOnErrorOriginal;

	boolean sourceHasThrownErrorPropagatingCommandException;
	boolean sourceHasRemovedRemoteConnection;

	Exception firstRunTestException;
	Exception firstRunVerifyException;

	// take any test that tests sending and receiving remote command

	public BroadcastReconnectionTest(BroadcastSetupHelper.TestWrapperWithEventLock test, boolean shouldDestroyFactory, boolean shouldRemoveConnectionOnError, BroadcastSetupHelper helper) {
	super(test);
	this.helper = helper;
	this.eventLock = test.getEventLock();

	this.shouldDestroyFactory = shouldDestroyFactory;
	this.shouldRemoveConnectionOnError = shouldRemoveConnectionOnError;
	setName("BroadcastReconnectionTest: shouldDestroyFactory = " + shouldDestroyFactory + "; shouldRemoveConnectionOnError = " + shouldRemoveConnectionOnError);
	setDescription("Invalidates RemoteConnection then removes it and runs cache sync test again");
	}

	protected TransportManager getTransportManager() {
	return ((AbstractSession)getSession()).getCommandManager().getTransportManager();
	}

	// Exception handler tries to re-creates local connection in a new thread.
	// Used only in case shouldRemoveConnectionOnError==true and
	// there is an exception thrown by local connection while it listens:
	// that happens in JMS case, doesn't happen with Oc4jJGroups.

	class LocalConnectionCreator implements ExceptionHandler, Runnable {
	// implements ExceptionHandler - used only in case shouldRemoveConnectionOnError==true
	AbstractSession session;
	ExceptionHandler originalExceptionHandler;
	boolean isActive;
	boolean hasReconnected;

	LocalConnectionCreator(AbstractSession session) {
	this.originalExceptionHandler = session.getExceptionHandler();
	session.setExceptionHandler(this);
	this.session = session;
	}

	@Override
	public Object handleException(RuntimeException exception) {
	if (exception instanceof RemoteCommandManagerException) {
	if (((RemoteCommandManagerException)exception).getErrorCode() == helper.getRcmExceptionErrorCodeOnFailureToCreateLocalConnection()) {
	if (isActive) {
	// already trying to reconnect - eat the exception.
	} else {
	isActive = true;
	hasReconnected = false;
	session.getServerPlatform().launchContainerRunnable(this);
	}
	return null;
	}
	}
	if (originalExceptionHandler != null) {
	return originalExceptionHandler.handleException(exception);
	} else {
	throw exception;
	}
	}

	@Override
	public void run() {
	while (isActive) {
	// no need for try block - we are eating the exception (see handleException method)
	session.getCommandManager().getTransportManager().createLocalConnection();
	if (session.getCommandManager().getTransportManager().getConnectionToLocalHost() != null) {
	// success!
	isActive = false;
	hasReconnected = true;
	}
	}
	}

	void clear() {
	if (isActive) {
	isActive = false;
	}
	session.setExceptionHandler(originalExceptionHandler);
	}
	}

	@Override
	protected void setup() throws Throwable {
	// save originals to restore back in reset
	shouldRemoveConnectionOnErrorOriginal = getTransportManager().shouldRemoveConnectionOnError();
	// set the new values for duration of the test
	getTransportManager().setShouldRemoveConnectionOnError(shouldRemoveConnectionOnError);
	// JMS only: if creation of local (listening) JMS connection fails
	// (which it will after the factory is destroyed) then keep trying to re-create listening connection.
	// Should succeed after the factory is recreated.
	// Note that there's no need for that for the main session (the one that sends messages):
	// the listening connection for it will be recreated together with the sending connection -
	// on attempt to send a message.
	if (shouldRemoveConnectionOnError && helper.isLocalConnectionRemovedOnListeningError()) {
	localConnectionCreators = new ArrayList();
	Iterator it = helper.getSessionsIterator();
	while (it.hasNext()) {
	AbstractSession session = (AbstractSession)it.next();
	if (session != getSession()) {
	localConnectionCreators.add(new LocalConnectionCreator(session));
	}
	}
	}
	// Shutdown the factory - after this sending and receiving remote commands should fail
	if (shouldDestroyFactory) {
	helper.destroyFactory();
	} else {
	stopFactory();
	}
	// disable the target listener.
	// need this in Oc4jJGroups case: the message sent through closed (or destroyed)
	// factory is still delivered to the target - and it happens before the
	// exception is thrown on the source.
	// EventLock would only accept the first unlocking event - and therefore unless
	// UNLOCKED_BY_TARGET_LISTENER is disabled exception (or connection removal)
	// would not show up as a EventLock's state after the test is complete.
	if (helper.shouldIgnoreTargetListenerInReconnectionTest()) {
	eventLock.disableState(BroadcastEventLock.UNLOCKED_BY_TARGET_LISTENER);
	}

	// setup the internal test for the first run
	super.setup();
	}

	@Override
	protected void test() throws Throwable {
	try {
	// This is expected to fail - the factory has been shut down.
	super.test();
	} catch (Exception ex) {
	// Because message published in a separate thread no exception is thrown in the main thread.
	// Exception is never thrown
	firstRunTestException = ex;
	}
	// Because the factory was either stopped or destroyed verify should fail.
	// However eventLock allows the wrapped test to proceed as soon as either
	// exception is thrown or connection is removed on the source - therefore there's no guarantee
	// that the message failed to be sent by the source will fail to be eventually recieved by the target -
	// may be if we waited a little bit more it would have reached the target.
	// It's not the case with OracleAQ-based JMS, but it seem to happen with Oc4jJGroups:
	// sending message throws exception on the source, but the message still reaches the target.
	try {
	super.verify();
	} catch (Exception ex) {
	// ignore
	firstRunVerifyException = ex;
	} finally {
	// reset the internal test - the first run is complete
	super.reset();
	}

	// look at the state of eventLock to see whether any events of interest have occurred during super.test()
	int state = eventLock.getState();

	// should be equal to !shouldRemoveConnectionOnError:
	// exception is thrown only in case connection is NOT to be removed on error
	sourceHasThrownErrorPropagatingCommandException = state == BroadcastEventLock.UNLOCKED_BY_SOURCE_EXCEPTION_HANDLER;
	// should be equal to shouldRemoveConnectionOnError:
	// connection is removed only in case connection is to be removed on error
	sourceHasRemovedRemoteConnection = state == BroadcastEventLock.UNLOCKED_BY_SOURCE_SESSION;

	// repair connections and recreate the factory in case the factory was destroyed
	if (shouldDestroyFactory) {
	// repair all connections, restart factory.
	resetConnections();
	} else {
	// restart the factory
	helper.startFactory();
	}

	if (shouldRemoveConnectionOnError \|\| shouldDestroyFactory) {
	// In External connection has been removed either as a result of the first run,
	// or by restConnection method,
	// therefore the second run will be immediately unlocked with
	// UNLOCKED_BY_SOURCE_SESSION unless we disable it.
	eventLock.disableState(BroadcastEventLock.UNLOCKED_BY_SOURCE_SESSION);
	}

	// now attempt to run the internal test again -
	// connections have been repaired therefore now it should pass.
	super.setup();
	super.test();
	}

	@Override
	protected void verify() throws Throwable {
	if (shouldRemoveConnectionOnError) {
	if (sourceHasThrownErrorPropagatingCommandException) {
	throw new TestErrorException("With shouldRemoveConnectionOnError==true there should've been NO ErrorPropagatingCommand RCMException thrown");
	}
	if (!sourceHasRemovedRemoteConnection) {
	throw new TestErrorException("With shouldRemoveConnectionOnError==true remote connection should've been removed");
	}
	} else {
	if (!sourceHasThrownErrorPropagatingCommandException) {
	throw new TestErrorException("With shouldRemoveConnectionOnError==false there should've been ErrorPropagatingCommand RCMException thrown");
	}
	if (sourceHasRemovedRemoteConnection) {
	throw new TestErrorException("With shouldRemoveConnectionOnError==false remote connection should've NOT been removed");
	}
	}
	if (firstRunTestException != null) {
	throw new TestErrorException("Unexpectedly firstRunTestException was thrown: ", firstRunTestException);
	}

	// look at the state of eventLock to see whether any events of interest have occurred during super.test()
	int state = eventLock.getState();
	if (helper.shouldIgnoreTargetListenerInReconnectionTest()) {
	// because targetListener is disabled, the successful result would unlock by timer -
	// that means there were no exceptions or connection removals.
	if (state != BroadcastEventLock.UNLOCKED_BY_TIMER) {
	throw new TestErrorException("Unexpected state " + state + " after the second run, BroadcastEventLock.UNLOCKED_BY_TIMER was expected");
	}
	} else {
	if (state != BroadcastEventLock.UNLOCKED_BY_TARGET_LISTENER) {
	throw new TestErrorException("Unexpected state " + state + " after the second run, BroadcastEventLock.UNLOCKED_BY_TARGET_LISTENER was expected");
	}
	if (firstRunVerifyException == null) {
	throw new TestErrorException("Unexpectedly firstRunVerifyException was NOT thrown");
	}
	}

	// verify the second run of internal test - now it should pass
	super.verify();
	}

	@Override
	public void reset() throws Throwable {
	if (localConnectionCreators != null) {
	Iterator it = localConnectionCreators.iterator();
	while (it.hasNext()) {
	LocalConnectionCreator localConnectionCreator = (LocalConnectionCreator)it.next();
	localConnectionCreator.clear();
	}
	localConnectionCreators = null;
	}
	if (shouldRemoveConnectionOnError) {
	// External connections for sessions other than main are still invalid - let's remove them.
	// Otherwise in case shouldRemoveConnectionOnError is set to false (either here or in another test)
	// the sessions will be stuck with the dead connections.
	helper.removeConnectionsForAllSessionsExcept((AbstractSession)getSession(), "external");
	}
	getTransportManager().setShouldRemoveConnectionOnError(shouldRemoveConnectionOnErrorOriginal);
	firstRunTestException = null;
	firstRunVerifyException = null;
	// some states could have been disabled - set back the original states' enabling.
	eventLock.enableAllStates();
	// reset internal test
	super.reset();
	}

	protected void stopFactory() throws Exception {
	helper.stopFactory();
	}

	protected void resetConnections() throws Exception {
	// the "main" session casted to AbstractSession
	AbstractSession abstractSession = (AbstractSession)getSession();

	// remove dead connections that haven't been already automatically removed.
	if (shouldRemoveConnectionOnError) {
	// getSession()'s external connection is already removed.
	// It will be automatically repaired next time it attempts sending a message.

	// External connections for all other sessions are still there:
	// they didn't send messages => haven't thrown an error => haven't been removed.
	// In these tests other sessions never send any messages, therefore we
	// don't really care about their external connections. However would the
	// other sessions attempt to send a message their invalid external connection
	// would throw exception and will be substituted with the new connection
	// (provided shouldRemoveConnectionOnError is still set to true).

	// remove local connections.
	if (helper.isLocalConnectionRemovedOnListeningError()) {
	// JMS case.
	// All local connections should be removed on listening error -
	// subscriber.receive method throwing JMSException. Nothing to do.
	} else {
	if (helper.isLocalConnectionAlsoExternalConnection()) {
	// Oc4jJGroups case - local connections for sessions other than the main one
	// should be removed.
	// The local connection is the same as external connection:
	// external connection for the main session was removed => local was removed, too.
	helper.removeConnectionsForAllSessionsExcept(abstractSession, "local");
	} else {
	// Currently never get here - provided for consistency only.
	helper.removeConnectionsForAllSessions("local");
	}
	}
	} else {
	// remove all connections for all sessions.
	helper.removeConnectionsForAllSessions("all");
	}
	// recreate the factory
	helper.createFactory();
	// start the factory
	helper.startFactory();

	// Re-create connections except those that will be re-created automatically
	// (those could be re-created "by hand", too - but that's not necessary).

	// No need to recreate external connections - they are automatically
	// re-creted on attempt to send a message.

	// Recreate local connections.
	if (localConnectionCreators != null) {
	// JMS shouldRemoveConnectionOnError==true case:
	// local connections on sessions other than main should be restored by exception handlers -
	// just wait until they are done.
	boolean allReconnected;
	do {
	allReconnected = true;
	Iterator it = localConnectionCreators.iterator();
	while (it.hasNext()) {
	LocalConnectionCreator localConnectionCreator = (LocalConnectionCreator)it.next();
	allReconnected = allReconnected && localConnectionCreator.hasReconnected;
	}
	if (!allReconnected) {
	Thread.sleep(1000);
	}
	} while (!allReconnected);
	} else {
	// for main session localConnection will be created simultaneously with the external one:
	// on attempt to send a message.
	helper.createConnectionsForAllSessionsExcept(abstractSession, "local");
	}
	}
	}