modules/persistence/src/test/java/org/apache/sandesha2/RMScenariosTest.java - sandesha - Git at Google

 /*
  * Copyright 2007 The Apache Software Foundation.
  * Copyright 2007 International Business Machines Corp.
  *
  * Licensed 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.apache.sandesha2;

 import java.io.File;
 import java.util.HashSet;
 import java.util.Iterator;
 import java.util.List;

 import junit.framework.AssertionFailedError;

 import org.apache.axis2.Constants;
 import org.apache.axis2.addressing.AddressingConstants;
 import org.apache.axis2.addressing.EndpointReference;
 import org.apache.axis2.client.Options;
 import org.apache.axis2.client.ServiceClient;
 import org.apache.axis2.context.ConfigurationContext;
 import org.apache.axis2.context.ConfigurationContextFactory;
 import org.apache.sandesha2.Sandesha2Constants;
 import org.apache.sandesha2.SandeshaTestCase;
 import org.apache.sandesha2.client.SandeshaClient;
 import org.apache.sandesha2.client.SandeshaClientConstants;
 import org.apache.sandesha2.client.SequenceReport;
 import org.apache.sandesha2.util.SandeshaUtil;

 public class RMScenariosTest extends SandeshaTestCase {

 	private boolean serverStarted = false;
 	protected ConfigurationContext configContext = null;

 	protected String to = "http://127.0.0.1:" + serverPort + "/axis2/services/RMSampleService";

 	protected String repoPath = "target" + File.separator + "repos" + File.separator + "persistence-server";
 	protected String axis2_xml = "target" + File.separator + "repos" + File.separator + "persistence-server" + File.separator + "server_axis2.xml";

 	protected String repoPathClient = "target" + File.separator + "repos" + File.separator + "persistence-client";
 	protected String axis2_xmlClient = "target" + File.separator + "repos" + File.separator + "persistence-client" + File.separator + "client_axis2.xml";

 	public RMScenariosTest() {
 		super("RMScenariosTest");
 	}

 	public RMScenariosTest(String name) {
 		super(name);
 	}

 	public void setUp() throws Exception {
 		super.setUp();

 		if (!serverStarted) {
 			startServer(repoPath, axis2_xml);
 			configContext = ConfigurationContextFactory.createConfigurationContextFromFileSystem(repoPathClient, axis2_xmlClient);
 		}
 		serverStarted = true;
 	}

 	/**
 	 * Override the teardown processing
 	 */
 	public void tearDown() throws Exception {
 		super.tearDown();
 	}

 	public void testPing() throws Exception {
 		// Run a ping test with sync acks
 		runPing(false, false);

 		// Run a ping test with async acks
 		runPing(true, true);
 	}

 	public void testAsyncEchoWithSyncAck() throws Exception {
 		// Test async echo with sync acks
 		Options clientOptions = new Options();
 		runEcho(clientOptions, true, false, false,true,true);
 	}

 	public void testAsyncEchoWithAsyncAck() throws Exception {
 		// Test async echo with async acks
 		Options clientOptions = new Options();
 		clientOptions = new Options();
 		runEcho(clientOptions, true, true, false,true,true);
 	}

 	public void testAsyncEchoWithAyncAckAndOffer() throws Exception {
 		// Test async echo with async acks and offer
 		Options clientOptions = new Options();
 		clientOptions = new Options();
 		clientOptions.setProperty(SandeshaClientConstants.OFFERED_SEQUENCE_ID,SandeshaUtil.getUUID());
 		runEcho(clientOptions, true, true, false,true,true);
 	}

 	public void testSyncEchoWithOffer() throws Exception {
 		// Test sync echo with an offer, and the 1.1 spec
 		Options clientOptions = new Options();
 		clientOptions.setProperty(SandeshaClientConstants.OFFERED_SEQUENCE_ID, SandeshaUtil.getUUID());
 		clientOptions.setProperty(SandeshaClientConstants.RM_SPEC_VERSION, Sandesha2Constants.SPEC_VERSIONS.v1_1);
 		runEcho(clientOptions, false, false, true, true, false);

 //		// Test sync echo with an offer, and the 1.0 spec. The offer is not automatic as this
 //		// is a client that hasn't been built from WSDL. If the user's operations had been
 //		// modelled properly then the offer would happen automatically.
 //		clientOptions = new Options();
 //		clientOptions.setProperty(SandeshaClientConstants.OFFERED_SEQUENCE_ID,SandeshaUtil.getUUID());
 //		clientOptions.setProperty(SandeshaClientConstants.RM_SPEC_VERSION,Sandesha2Constants.SPEC_VERSIONS.v1_0);
 //		runEcho(clientOptions, false, false, true,false,false);
 	}

 	public void testSyncEcho() throws Exception {
 		// Test sync echo with no offer, and the 1.1 spec
 		Options clientOptions = new Options();
 		clientOptions.setProperty(SandeshaClientConstants.RM_SPEC_VERSION, Sandesha2Constants.SPEC_VERSIONS.v1_1);
 		runEcho(clientOptions, false, false, true, true, true);
 	}

 	public void runPing(boolean asyncAcks, boolean stopListener) throws Exception {

 		Options clientOptions = new Options();

 		ServiceClient serviceClient = new ServiceClient(configContext, null);
 		serviceClient.setOptions(clientOptions);

 		String sequenceKey = SandeshaUtil.getUUID();

 		clientOptions.setAction(pingAction);
 		clientOptions.setTo(new EndpointReference(to));
 		clientOptions.setProperty(SandeshaClientConstants.SEQUENCE_KEY, sequenceKey);
 		clientOptions.setProperty(SandeshaClientConstants.LAST_MESSAGE, Constants.VALUE_TRUE);

 		if (asyncAcks) {
 			clientOptions.setUseSeparateListener(true);
 		}

 		serviceClient.fireAndForget(getPingOMBlock("ping1"));

 		long limit = System.currentTimeMillis() + waitTime;
 		Error lastError = null;
 		while (System.currentTimeMillis() < limit) {
 			Thread.sleep(tickTime); // Try the assertions each tick interval, until they pass or we time out

 			try {
 				SequenceReport sequenceReport = SandeshaClient.getOutgoingSequenceReport(serviceClient);
 				System.out.println("Checking Outbound Sequence: " + sequenceReport.getSequenceID());
 				assertTrue("Checking completed messages", sequenceReport.getCompletedMessages().contains(new Long(1)));
 				assertEquals("Checking sequence terminated", SequenceReport.SEQUENCE_STATUS_TERMINATED, sequenceReport.getSequenceStatus());
 				assertEquals("Checking sequence direction", SequenceReport.SEQUENCE_DIRECTION_OUT, sequenceReport.getSequenceDirection());

 				lastError = null;
 				break;
 			} catch (Error e) {
 				e.printStackTrace();
 				System.out.println("Possible error:" + e);
 				lastError = e;
 			}
 		}

 		if (lastError != null) throw lastError;

 		if (stopListener)
 			configContext.getListenerManager().stop();

 		serviceClient.cleanup();

 	}

 	public void runEcho(Options clientOptions, boolean asyncReply, boolean asyncAcks, boolean explicitTermination, boolean checkInboundTermination, boolean stopListener) throws Exception {

 		String sequenceKey = SandeshaUtil.getUUID();

 		ServiceClient serviceClient = new ServiceClient(configContext, null);
 		serviceClient.setOptions(clientOptions);

 		clientOptions.setAction(echoAction);
 		clientOptions.setTo(new EndpointReference(to));
 		clientOptions.setProperty(SandeshaClientConstants.SEQUENCE_KEY, sequenceKey);
 		clientOptions.setTransportInProtocol(Constants.TRANSPORT_HTTP);

 		if (asyncReply || asyncAcks) {
 			clientOptions.setUseSeparateListener(true);

 			if (asyncAcks) {
 				String acksTo = serviceClient.getMyEPR(Constants.TRANSPORT_HTTP).getAddress();
 				clientOptions.setProperty(SandeshaClientConstants.AcksTo, acksTo);
 			} else {
 				String acksTo = AddressingConstants.Final.WSA_ANONYMOUS_URL;
 				clientOptions.setProperty(SandeshaClientConstants.AcksTo, acksTo);
 			}
 		}

 		if (asyncAcks) {
 			String acksTo = serviceClient.getMyEPR(Constants.TRANSPORT_HTTP).getAddress();
 			clientOptions.setProperty(SandeshaClientConstants.AcksTo, acksTo);
 		} else {
 			String acksTo = AddressingConstants.Final.WSA_ANONYMOUS_URL;
 			clientOptions.setProperty(SandeshaClientConstants.AcksTo, acksTo);
 		}

 		// Establish a baseline count for inbound sequences
 		List<SequenceReport> oldIncomingReports = SandeshaClient.getIncomingSequenceReports(configContext);

 		TestCallback callback1 = new TestCallback("Callback 1");
 		serviceClient.sendReceiveNonBlocking(getEchoOMBlock("echo1", sequenceKey), callback1);

 		TestCallback callback2 = new TestCallback("Callback 2");
 		serviceClient.sendReceiveNonBlocking(getEchoOMBlock("echo2", sequenceKey), callback2);

 		if (!explicitTermination
 				&&
 				!Sandesha2Constants.SPEC_VERSIONS.v1_1.equals(clientOptions.getProperty(SandeshaClientConstants.RM_SPEC_VERSION)))
 		{

 			clientOptions.setProperty(SandeshaClientConstants.LAST_MESSAGE, Constants.VALUE_TRUE);
 		}

 		TestCallback callback3 = new TestCallback("Callback 3");
 		serviceClient.sendReceiveNonBlocking(getEchoOMBlock("echo3", sequenceKey), callback3);

 		if (explicitTermination) {
 			Thread.sleep(10000);
 			SandeshaClient.terminateSequence(serviceClient);
 		}

 		long limit = System.currentTimeMillis() + waitTime;
 		Error lastError = null;
 		while (System.currentTimeMillis() < limit) {
 			Thread.sleep(tickTime); // Try the assertions each tick interval, until they pass or we time out

 			try {

 				//assertions for the out sequence.
 				SequenceReport outgoingSequenceReport = SandeshaClient.getOutgoingSequenceReport(serviceClient);
 				System.out.println("Checking Outbound Sequence: " + outgoingSequenceReport.getSequenceID());
 				assertTrue("Outbound message #1", outgoingSequenceReport.getCompletedMessages().contains(new Long(1)));
 				assertTrue("Outbound message #2", outgoingSequenceReport.getCompletedMessages().contains(new Long(2)));
 				assertTrue("Outbound message #3", outgoingSequenceReport.getCompletedMessages().contains(new Long(3)));
 				assertEquals("Outbound sequence status: TERMINATED", SequenceReport.SEQUENCE_STATUS_TERMINATED, outgoingSequenceReport.getSequenceStatus());
 				assertEquals("Outbound sequence direction: OUT", SequenceReport.SEQUENCE_DIRECTION_OUT, outgoingSequenceReport.getSequenceDirection());

 				//assertions for the inbound sequence. The one we care about is a new sequence,
 				//so it will not exist in the oldSequences list.
 				List<SequenceReport> incomingSequences = SandeshaClient.getIncomingSequenceReports(configContext);
 				SequenceReport incomingSequenceReport = getNewReport(incomingSequences, oldIncomingReports);
 				System.out.println("Checking Inbound Sequence: " + incomingSequenceReport.getSequenceID());
 				String offer = (String) clientOptions.getProperty(SandeshaClientConstants.OFFERED_SEQUENCE_ID);
 				if (offer != null) assertEquals("Inbound seq id", offer, incomingSequenceReport.getSequenceID());
 				assertEquals("Inbound message count", 3, incomingSequenceReport.getCompletedMessages().size());
 				assertTrue("Inbound message #1", incomingSequenceReport.getCompletedMessages().contains(new Long(1)));
 				assertTrue("Inbound message #2", incomingSequenceReport.getCompletedMessages().contains(new Long(2)));
 				assertTrue("Inbound message #3", incomingSequenceReport.getCompletedMessages().contains(new Long(3)));

 				if (checkInboundTermination)
 					assertEquals("Inbound sequence status: TERMINATED", SequenceReport.SEQUENCE_STATUS_TERMINATED, incomingSequenceReport.getSequenceStatus());

 				assertEquals("Inbound sequence direction: IN", SequenceReport.SEQUENCE_DIRECTION_IN, incomingSequenceReport.getSequenceDirection());

 				assertTrue("Callback #1", callback1.isComplete());
 				assertEquals("Callback #1 data", "echo1", callback1.getResult());

 				assertTrue("Callback #2", callback2.isComplete());
 				assertEquals("Callback #2 data", "echo1echo2", callback2.getResult());

 				assertTrue("Callback #3", callback3.isComplete());
 				assertEquals("Callback #3 data", "echo1echo2echo3", callback3.getResult());

 				lastError = null;
 				break;
 			} catch (Error e) {
 				e.printStackTrace();
 				System.out.println("Possible error:" + e);
 				lastError = e;
 			}
 		}
 		if (lastError != null) throw lastError;

 		if (stopListener)
 			configContext.getListenerManager().stop();

 		serviceClient.cleanup();
 	}

 	// Scan through lists of old and new incoming sequences, to find the sequence that
 	// was established by this test. Note that some of the old sequences may have timed out.
 	private SequenceReport getNewReport(List<SequenceReport> incomingSequences, List<SequenceReport> oldIncomingReports) {
 		HashSet<String> sequenceIds = new HashSet<String>();
 		for (Iterator<SequenceReport> oldSequences = oldIncomingReports.iterator(); oldSequences.hasNext();) {
 			SequenceReport report = (SequenceReport) oldSequences.next();
 			sequenceIds.add(report.getSequenceID());
 		}
 		for (Iterator<SequenceReport> currentSequences = incomingSequences.iterator(); currentSequences.hasNext();) {
 			SequenceReport report = (SequenceReport) currentSequences.next();
 			if (!sequenceIds.contains(report.getSequenceID())) {
 				return report;
 			}
 		}
 		throw new AssertionFailedError("Failed to find a new reply sequence");
 	}


 }
	/*
	* Copyright 2007 The Apache Software Foundation.
	* Copyright 2007 International Business Machines Corp.
	*
	* Licensed 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.apache.sandesha2;

	import java.io.File;
	import java.util.HashSet;
	import java.util.Iterator;
	import java.util.List;

	import junit.framework.AssertionFailedError;

	import org.apache.axis2.Constants;
	import org.apache.axis2.addressing.AddressingConstants;
	import org.apache.axis2.addressing.EndpointReference;
	import org.apache.axis2.client.Options;
	import org.apache.axis2.client.ServiceClient;
	import org.apache.axis2.context.ConfigurationContext;
	import org.apache.axis2.context.ConfigurationContextFactory;
	import org.apache.sandesha2.Sandesha2Constants;
	import org.apache.sandesha2.SandeshaTestCase;
	import org.apache.sandesha2.client.SandeshaClient;
	import org.apache.sandesha2.client.SandeshaClientConstants;
	import org.apache.sandesha2.client.SequenceReport;
	import org.apache.sandesha2.util.SandeshaUtil;

	public class RMScenariosTest extends SandeshaTestCase {

	private boolean serverStarted = false;
	protected ConfigurationContext configContext = null;

	protected String to = "http://127.0.0.1:" + serverPort + "/axis2/services/RMSampleService";

	protected String repoPath = "target" + File.separator + "repos" + File.separator + "persistence-server";
	protected String axis2_xml = "target" + File.separator + "repos" + File.separator + "persistence-server" + File.separator + "server_axis2.xml";

	protected String repoPathClient = "target" + File.separator + "repos" + File.separator + "persistence-client";
	protected String axis2_xmlClient = "target" + File.separator + "repos" + File.separator + "persistence-client" + File.separator + "client_axis2.xml";

	public RMScenariosTest() {
	super("RMScenariosTest");
	}

	public RMScenariosTest(String name) {
	super(name);
	}

	public void setUp() throws Exception {
	super.setUp();

	if (!serverStarted) {
	startServer(repoPath, axis2_xml);
	configContext = ConfigurationContextFactory.createConfigurationContextFromFileSystem(repoPathClient, axis2_xmlClient);
	}
	serverStarted = true;
	}

	/**
	* Override the teardown processing
	*/
	public void tearDown() throws Exception {
	super.tearDown();
	}

	public void testPing() throws Exception {
	// Run a ping test with sync acks
	runPing(false, false);

	// Run a ping test with async acks
	runPing(true, true);
	}

	public void testAsyncEchoWithSyncAck() throws Exception {
	// Test async echo with sync acks
	Options clientOptions = new Options();
	runEcho(clientOptions, true, false, false,true,true);
	}

	public void testAsyncEchoWithAsyncAck() throws Exception {
	// Test async echo with async acks
	Options clientOptions = new Options();
	clientOptions = new Options();
	runEcho(clientOptions, true, true, false,true,true);
	}

	public void testAsyncEchoWithAyncAckAndOffer() throws Exception {
	// Test async echo with async acks and offer
	Options clientOptions = new Options();
	clientOptions = new Options();
	clientOptions.setProperty(SandeshaClientConstants.OFFERED_SEQUENCE_ID,SandeshaUtil.getUUID());
	runEcho(clientOptions, true, true, false,true,true);
	}

	public void testSyncEchoWithOffer() throws Exception {
	// Test sync echo with an offer, and the 1.1 spec
	Options clientOptions = new Options();
	clientOptions.setProperty(SandeshaClientConstants.OFFERED_SEQUENCE_ID, SandeshaUtil.getUUID());
	clientOptions.setProperty(SandeshaClientConstants.RM_SPEC_VERSION, Sandesha2Constants.SPEC_VERSIONS.v1_1);
	runEcho(clientOptions, false, false, true, true, false);

	// // Test sync echo with an offer, and the 1.0 spec. The offer is not automatic as this
	// // is a client that hasn't been built from WSDL. If the user's operations had been
	// // modelled properly then the offer would happen automatically.
	// clientOptions = new Options();
	// clientOptions.setProperty(SandeshaClientConstants.OFFERED_SEQUENCE_ID,SandeshaUtil.getUUID());
	// clientOptions.setProperty(SandeshaClientConstants.RM_SPEC_VERSION,Sandesha2Constants.SPEC_VERSIONS.v1_0);
	// runEcho(clientOptions, false, false, true,false,false);
	}

	public void testSyncEcho() throws Exception {
	// Test sync echo with no offer, and the 1.1 spec
	Options clientOptions = new Options();
	clientOptions.setProperty(SandeshaClientConstants.RM_SPEC_VERSION, Sandesha2Constants.SPEC_VERSIONS.v1_1);
	runEcho(clientOptions, false, false, true, true, true);
	}

	public void runPing(boolean asyncAcks, boolean stopListener) throws Exception {

	Options clientOptions = new Options();

	ServiceClient serviceClient = new ServiceClient(configContext, null);
	serviceClient.setOptions(clientOptions);

	String sequenceKey = SandeshaUtil.getUUID();

	clientOptions.setAction(pingAction);
	clientOptions.setTo(new EndpointReference(to));
	clientOptions.setProperty(SandeshaClientConstants.SEQUENCE_KEY, sequenceKey);
	clientOptions.setProperty(SandeshaClientConstants.LAST_MESSAGE, Constants.VALUE_TRUE);

	if (asyncAcks) {
	clientOptions.setUseSeparateListener(true);
	}

	serviceClient.fireAndForget(getPingOMBlock("ping1"));

	long limit = System.currentTimeMillis() + waitTime;
	Error lastError = null;
	while (System.currentTimeMillis() < limit) {
	Thread.sleep(tickTime); // Try the assertions each tick interval, until they pass or we time out

	try {
	SequenceReport sequenceReport = SandeshaClient.getOutgoingSequenceReport(serviceClient);
	System.out.println("Checking Outbound Sequence: " + sequenceReport.getSequenceID());
	assertTrue("Checking completed messages", sequenceReport.getCompletedMessages().contains(new Long(1)));
	assertEquals("Checking sequence terminated", SequenceReport.SEQUENCE_STATUS_TERMINATED, sequenceReport.getSequenceStatus());
	assertEquals("Checking sequence direction", SequenceReport.SEQUENCE_DIRECTION_OUT, sequenceReport.getSequenceDirection());

	lastError = null;
	break;
	} catch (Error e) {
	e.printStackTrace();
	System.out.println("Possible error:" + e);
	lastError = e;
	}
	}

	if (lastError != null) throw lastError;

	if (stopListener)
	configContext.getListenerManager().stop();

	serviceClient.cleanup();

	}

	public void runEcho(Options clientOptions, boolean asyncReply, boolean asyncAcks, boolean explicitTermination, boolean checkInboundTermination, boolean stopListener) throws Exception {

	String sequenceKey = SandeshaUtil.getUUID();

	ServiceClient serviceClient = new ServiceClient(configContext, null);
	serviceClient.setOptions(clientOptions);

	clientOptions.setAction(echoAction);
	clientOptions.setTo(new EndpointReference(to));
	clientOptions.setProperty(SandeshaClientConstants.SEQUENCE_KEY, sequenceKey);
	clientOptions.setTransportInProtocol(Constants.TRANSPORT_HTTP);

	if (asyncReply \|\| asyncAcks) {
	clientOptions.setUseSeparateListener(true);

	if (asyncAcks) {
	String acksTo = serviceClient.getMyEPR(Constants.TRANSPORT_HTTP).getAddress();
	clientOptions.setProperty(SandeshaClientConstants.AcksTo, acksTo);
	} else {
	String acksTo = AddressingConstants.Final.WSA_ANONYMOUS_URL;
	clientOptions.setProperty(SandeshaClientConstants.AcksTo, acksTo);
	}
	}

	if (asyncAcks) {
	String acksTo = serviceClient.getMyEPR(Constants.TRANSPORT_HTTP).getAddress();
	clientOptions.setProperty(SandeshaClientConstants.AcksTo, acksTo);
	} else {
	String acksTo = AddressingConstants.Final.WSA_ANONYMOUS_URL;
	clientOptions.setProperty(SandeshaClientConstants.AcksTo, acksTo);
	}

	// Establish a baseline count for inbound sequences
	List<SequenceReport> oldIncomingReports = SandeshaClient.getIncomingSequenceReports(configContext);

	TestCallback callback1 = new TestCallback("Callback 1");
	serviceClient.sendReceiveNonBlocking(getEchoOMBlock("echo1", sequenceKey), callback1);

	TestCallback callback2 = new TestCallback("Callback 2");
	serviceClient.sendReceiveNonBlocking(getEchoOMBlock("echo2", sequenceKey), callback2);

	if (!explicitTermination
	&&
	!Sandesha2Constants.SPEC_VERSIONS.v1_1.equals(clientOptions.getProperty(SandeshaClientConstants.RM_SPEC_VERSION)))
	{

	clientOptions.setProperty(SandeshaClientConstants.LAST_MESSAGE, Constants.VALUE_TRUE);
	}

	TestCallback callback3 = new TestCallback("Callback 3");
	serviceClient.sendReceiveNonBlocking(getEchoOMBlock("echo3", sequenceKey), callback3);

	if (explicitTermination) {
	Thread.sleep(10000);
	SandeshaClient.terminateSequence(serviceClient);
	}

	long limit = System.currentTimeMillis() + waitTime;
	Error lastError = null;
	while (System.currentTimeMillis() < limit) {
	Thread.sleep(tickTime); // Try the assertions each tick interval, until they pass or we time out

	try {

	//assertions for the out sequence.
	SequenceReport outgoingSequenceReport = SandeshaClient.getOutgoingSequenceReport(serviceClient);
	System.out.println("Checking Outbound Sequence: " + outgoingSequenceReport.getSequenceID());
	assertTrue("Outbound message #1", outgoingSequenceReport.getCompletedMessages().contains(new Long(1)));
	assertTrue("Outbound message #2", outgoingSequenceReport.getCompletedMessages().contains(new Long(2)));
	assertTrue("Outbound message #3", outgoingSequenceReport.getCompletedMessages().contains(new Long(3)));
	assertEquals("Outbound sequence status: TERMINATED", SequenceReport.SEQUENCE_STATUS_TERMINATED, outgoingSequenceReport.getSequenceStatus());
	assertEquals("Outbound sequence direction: OUT", SequenceReport.SEQUENCE_DIRECTION_OUT, outgoingSequenceReport.getSequenceDirection());

	//assertions for the inbound sequence. The one we care about is a new sequence,
	//so it will not exist in the oldSequences list.
	List<SequenceReport> incomingSequences = SandeshaClient.getIncomingSequenceReports(configContext);
	SequenceReport incomingSequenceReport = getNewReport(incomingSequences, oldIncomingReports);
	System.out.println("Checking Inbound Sequence: " + incomingSequenceReport.getSequenceID());
	String offer = (String) clientOptions.getProperty(SandeshaClientConstants.OFFERED_SEQUENCE_ID);
	if (offer != null) assertEquals("Inbound seq id", offer, incomingSequenceReport.getSequenceID());
	assertEquals("Inbound message count", 3, incomingSequenceReport.getCompletedMessages().size());
	assertTrue("Inbound message #1", incomingSequenceReport.getCompletedMessages().contains(new Long(1)));
	assertTrue("Inbound message #2", incomingSequenceReport.getCompletedMessages().contains(new Long(2)));
	assertTrue("Inbound message #3", incomingSequenceReport.getCompletedMessages().contains(new Long(3)));

	if (checkInboundTermination)
	assertEquals("Inbound sequence status: TERMINATED", SequenceReport.SEQUENCE_STATUS_TERMINATED, incomingSequenceReport.getSequenceStatus());

	assertEquals("Inbound sequence direction: IN", SequenceReport.SEQUENCE_DIRECTION_IN, incomingSequenceReport.getSequenceDirection());

	assertTrue("Callback #1", callback1.isComplete());
	assertEquals("Callback #1 data", "echo1", callback1.getResult());

	assertTrue("Callback #2", callback2.isComplete());
	assertEquals("Callback #2 data", "echo1echo2", callback2.getResult());

	assertTrue("Callback #3", callback3.isComplete());
	assertEquals("Callback #3 data", "echo1echo2echo3", callback3.getResult());

	lastError = null;
	break;
	} catch (Error e) {
	e.printStackTrace();
	System.out.println("Possible error:" + e);
	lastError = e;
	}
	}
	if (lastError != null) throw lastError;

	if (stopListener)
	configContext.getListenerManager().stop();

	serviceClient.cleanup();
	}

	// Scan through lists of old and new incoming sequences, to find the sequence that
	// was established by this test. Note that some of the old sequences may have timed out.
	private SequenceReport getNewReport(List<SequenceReport> incomingSequences, List<SequenceReport> oldIncomingReports) {
	HashSet<String> sequenceIds = new HashSet<String>();
	for (Iterator<SequenceReport> oldSequences = oldIncomingReports.iterator(); oldSequences.hasNext();) {
	SequenceReport report = (SequenceReport) oldSequences.next();
	sequenceIds.add(report.getSequenceID());
	}
	for (Iterator<SequenceReport> currentSequences = incomingSequences.iterator(); currentSequences.hasNext();) {
	SequenceReport report = (SequenceReport) currentSequences.next();
	if (!sequenceIds.contains(report.getSequenceID())) {
	return report;
	}
	}
	throw new AssertionFailedError("Failed to find a new reply sequence");
	}


	}