svn commit: r1142350 [10/18] - in /incubator/airavata/donations/ogce-donation/modules/ws-messaging: broker/ broker/customLibs/ broker/customLibs/activeMQ/ broker/src/ broker/src/main/ broker/src/main/java/ broker/src/main/java/wsmg/ broker/src/main/jav...

[sm...@apache.org]

Added: incubator/airavata/donations/ogce-donation/modules/ws-messaging/broker/src/test/java/performance_evaluator/rtt/PerformanceTest.java
URL: http://svn.apache.org/viewvc/incubator/airavata/donations/ogce-donation/modules/ws-messaging/broker/src/test/java/performance_evaluator/rtt/PerformanceTest.java?rev=1142350&view=auto
==============================================================================
--- incubator/airavata/donations/ogce-donation/modules/ws-messaging/broker/src/test/java/performance_evaluator/rtt/PerformanceTest.java (added)
+++ incubator/airavata/donations/ogce-donation/modules/ws-messaging/broker/src/test/java/performance_evaluator/rtt/PerformanceTest.java Sun Jul  3 00:29:24 2011
@@ -0,0 +1,436 @@
+package performance_evaluator.rtt;
+
+import java.io.BufferedWriter;
+import java.io.File;
+import java.io.FileNotFoundException;
+import java.io.FileWriter;
+import java.io.IOException;
+import java.io.Writer;
+import java.net.URL;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Iterator;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Properties;
+import java.util.concurrent.CountDownLatch;
+
+import performance_evaluator.rtt.util.ConfigKeys;
+import performance_evaluator.rtt.util.LoadMsgPayload;
+import performance_evaluator.rtt.util.LoadXpath;
+
+/**
+ * @author dinesh
+ */
+
+class Stat {
+	String name;
+	Object value;
+
+	public Stat(String k, Object v) {
+		name = k;
+		value = v;
+	}
+
+}
+
+public class PerformanceTest {
+
+	public static int NOTIFICATIONS_PUBLISHED_PER_TOPIC = 0;
+	static String payload = null;
+	static LinkedList<String> xpathList = null;
+	public static long totalRoundTripTime = 0;
+	private static long avgRountTripTime = 0;
+	public static BufferedWriter out = null;
+	public static Properties configurations = null;
+	public static long avgPublishRTTime = 0l;
+	private static long totalPublishRTT = 0l;
+	private static int notifPerTopic = 0;
+	private static int noTopicsPublished = 0;
+	private static String protocol = "";
+	private static int payLoadMultiplier = 1;
+	private static int consumerPort = 3345;
+	private static long testExpirationTime = 0l;
+	private static int numberOfSubscriber = 0;
+	private static int numMultiThreadsSupportPerSub = 0;
+	private static String topicPrefix = "";
+
+	public static void main(String[] args) throws Exception {
+		loadConfigurationsFromFile();
+		testPerformance();
+	}
+
+	public static void testPerformance() throws Exception {
+
+		setConfigurationValues();
+		File outfile = new File("performance.log");
+		CountDownLatch publiserhStartSignal = new CountDownLatch(1);
+		CountDownLatch publisherDoneSignal = new CountDownLatch(
+				noTopicsPublished);
+		NotificationManager notifManagerArray[] = new NotificationManager[numberOfSubscriber];
+		StatCalculatorThread statCalcThread[] = new StatCalculatorThread[numberOfSubscriber];
+		setPayload(payLoadMultiplier);
+
+		for (int j = 0; j < numberOfSubscriber; j++) {
+			notifManagerArray[j] = new NotificationManager(configurations
+					.getProperty(ConfigKeys.BROKER_URL), consumerPort + j,
+					protocol, numMultiThreadsSupportPerSub);
+
+		}
+
+		// thread to calculate stats for notification manager
+		// set the subscriptions depending on the topic or xpath based
+		int arrayIndex = 0;
+		int totalReceivers = 0;
+		createSubscriberArray(noTopicsPublished, numberOfSubscriber,
+				notifManagerArray, arrayIndex);
+		System.out
+				.println("subscribing to topics completed, creating publisher threads");
+
+		// start publishers
+		PublisherThread[] publisher = new PublisherThread[noTopicsPublished];
+		createPublishers(noTopicsPublished, protocol, publiserhStartSignal,
+				publisherDoneSignal, publisher);
+		System.out.println("sending signal to start publishing...");
+		long publisherStartTime = System.currentTimeMillis();
+		long startTime = System.currentTimeMillis();
+		publiserhStartSignal.countDown(); // let all threads proceed
+		
+
+		for (int j = 0; j < numberOfSubscriber; j++) {
+			statCalcThread[j] = new StatCalculatorThread(notifManagerArray[j],
+					testExpirationTime);
+			statCalcThread[j].start();
+		}
+
+		
+		publisherDoneSignal.await(); // wait for all to finish
+
+		for (int j = 0; j < noTopicsPublished; j++) {
+			totalPublishRTT += publisher[j].getAvgPubTime();
+		}
+
+		avgPublishRTTime = totalPublishRTT / noTopicsPublished;
+		long publishersRunningTime = System.currentTimeMillis()
+				- publisherStartTime;
+		System.out.println("finished publishing messgaes.");
+
+		for (StatCalculatorThread stats : statCalcThread) {
+			stats.join();
+		}
+
+		long stopTime = 0l;
+		long totNumberOfMessagesReceived = 0;
+
+		for (StatCalculatorThread stats : statCalcThread) {
+			stopTime = stopTime < stats.getLastMsgReceivedTime() ? stats
+					.getLastMsgReceivedTime() : stopTime;
+			totalRoundTripTime += stats.getTotalTime();
+			totNumberOfMessagesReceived += stats.getNumberOfMsgReceived();
+		}
+
+		for (NotificationManager notifMngr : notifManagerArray) {
+			totalReceivers += notifMngr.getNoTopicsSubscribed();
+		}
+
+		avgRountTripTime = totalRoundTripTime / totNumberOfMessagesReceived;
+		long executionTime = stopTime - startTime;
+		double throughtput = (totNumberOfMessagesReceived * 1000)
+				/ (executionTime);
+
+		List<Stat> statistics = new ArrayList<Stat>();
+
+		statistics.add(new Stat("Payload size (bytes)", payload
+				.getBytes("US-ASCII").length));
+		statistics.add(new Stat("Protocol", protocol));
+		statistics.add(new Stat("# total expected Msgs", totalReceivers
+				* notifPerTopic));
+		statistics.add(new Stat("# total msgs received",
+				totNumberOfMessagesReceived));
+		setStatList(notifPerTopic, noTopicsPublished, publishersRunningTime,
+				executionTime, throughtput, statistics);
+		printStatistics(statistics, outfile);
+
+		for (NotificationManager notifMngr : notifManagerArray) {
+			notifMngr.cleanup();
+		}
+
+		System.out.println("end of test");
+		System.exit(0);
+	}
+
+	private static void setConfigurationValues() {
+		notifPerTopic = Integer.parseInt(configurations
+				.getProperty(ConfigKeys.NOTIFICATIONS_PUBLISHED_PER_TOPIC));
+		noTopicsPublished = Integer.parseInt(configurations
+				.getProperty(ConfigKeys.NUMBER_OF_TOPICS_PUBLISHED));
+		protocol = configurations.getProperty(ConfigKeys.PROTOCOL);
+		payLoadMultiplier = Integer.parseInt(configurations
+				.getProperty(ConfigKeys.PAYLOAD_MULTIPLYER));
+		consumerPort = Integer.parseInt(configurations
+				.getProperty(ConfigKeys.CONSUMER_PORT));
+		testExpirationTime = Math.max(20000, Long.parseLong(configurations
+				.getProperty(ConfigKeys.PERFORMANCE_TEST_TIMEOUT, "20000")));
+		numberOfSubscriber = Integer.parseInt(configurations
+				.getProperty(ConfigKeys.NUMBER_OF_SUBSCRIBERS));
+		numMultiThreadsSupportPerSub = Integer.parseInt(configurations
+				.getProperty(ConfigKeys.MULTI_THREAD_PER_SUB));
+		topicPrefix = "topic"
+				+ configurations.getProperty(ConfigKeys.TOPIC_SIMPLE);
+		NOTIFICATIONS_PUBLISHED_PER_TOPIC = notifPerTopic;
+	}
+
+	private static void setStatList(int notifPerTopic, int noTopicsPublished,
+			long publishersRunningTime, long executionTime, double throughtput,
+			List<Stat> statistics) {
+
+		statistics.add(new Stat("# topics published", noTopicsPublished));
+		statistics.add(new Stat("Total RTT (millis)", totalRoundTripTime));
+		statistics.add(new Stat("Average RTT (millis)", avgRountTripTime));
+		statistics.add(new Stat("Total published to receive time (millis)",
+				executionTime));
+		statistics
+				.add(new Stat("Throughput (messages per second)", throughtput));
+		statistics.add(new Stat("Total publish RTT (millis)", totalPublishRTT));
+		statistics.add(new Stat("Average publish RTT (millis)",
+				avgPublishRTTime));
+		statistics.add(new Stat("publisher duration (millis)",
+				publishersRunningTime));
+		statistics.add(new Stat("Publisher throughput (messages per second)",
+				noTopicsPublished * notifPerTopic * 1000
+						/ publishersRunningTime));
+	}
+
+	private static void setPayload(int payLoadMultiplier) {
+		String tempPayload = "";
+		try {
+			tempPayload = LoadMsgPayload.getInstance().getMessage("payload.txt");
+		} catch (FileNotFoundException e2) {
+			e2.printStackTrace();
+		} catch (IOException e) {
+			e.printStackTrace();
+		}
+
+		for (int i = 1; i <= payLoadMultiplier; i++) {
+			payload += tempPayload;
+		}
+	}
+
+	private static void createPublishers(int noTopicsPublished,
+			String protocol, CountDownLatch publiserhStartSignal,
+			CountDownLatch publisherDoneSignal, PublisherThread[] publisher) {
+		int threadId = 0;
+		for (int j = 0; j < noTopicsPublished; j++) {
+			threadId++;
+			publisher[j] = new PublisherThread(protocol, configurations
+					.getProperty(ConfigKeys.BROKER_URL), topicPrefix + j,
+					payload, publiserhStartSignal, publisherDoneSignal,
+					threadId);
+			publisher[j].start();
+		}
+	}
+
+	private static void createSubscriberArray(int noTopicsPublished,
+			int numberOfSubscriber, NotificationManager[] notifManagerArray,
+			int arrayIndex) throws Exception, IOException {
+		if ("false".equalsIgnoreCase(configurations
+				.getProperty(ConfigKeys.IS_XPATH_ENABLED))) {
+			if (numberOfSubscriber <= noTopicsPublished) {
+				for (int i = 0; i < noTopicsPublished; ++i) {
+					notifManagerArray[arrayIndex]
+							.createTopicSubscription(topicPrefix + i);
+					notifManagerArray[arrayIndex++].incNoTopicsSubscribed();
+					if (arrayIndex >= numberOfSubscriber) {
+						arrayIndex = 0;
+					}
+				}
+			} else {
+				int topicIndex = 0;
+				for (int i = 0; i < numberOfSubscriber; ++i) {
+					notifManagerArray[i].createTopicSubscription(topicPrefix
+							+ topicIndex++);
+					notifManagerArray[i].incNoTopicsSubscribed();
+					if (topicIndex >= noTopicsPublished) {
+						topicIndex = 0;
+					}
+				}
+			}
+		} else {
+			xpathList = LoadXpath.getInstace().getXpathList("xpath.list");
+			if (numberOfSubscriber <= noTopicsPublished) {
+				Iterator<String> ite = xpathList.iterator();
+				for (int i = 0; i < noTopicsPublished; ++i) {
+					if (!ite.hasNext())
+						ite = xpathList.iterator();
+
+					notifManagerArray[arrayIndex].createXpathSubscription(
+							topicPrefix + i, ite.next());
+					notifManagerArray[arrayIndex++].incNoTopicsSubscribed();
+					if (arrayIndex >= numberOfSubscriber) {
+						arrayIndex = 0;
+					}
+				}
+			} else {
+				int topicIndex = 0;
+				for (int i = 0; i < numberOfSubscriber; ++i) {
+					notifManagerArray[i].incNoTopicsSubscribed();
+					if (topicIndex >= noTopicsPublished) {
+						topicIndex = 0;
+					}
+				}
+			}
+		}
+	}
+
+	private static Properties getDefaults() {
+		Properties defaults = new Properties();
+		defaults.setProperty(ConfigKeys.BROKER_URL,
+				"http://localhost:8080/axis2/services/EventingService");
+		defaults.setProperty(ConfigKeys.TOPIC_SIMPLE, "simpleSampleTopic");
+		defaults.setProperty(ConfigKeys.CONSUMER_PORT, "6666");
+		defaults.setProperty(ConfigKeys.NOTIFICATIONS_PUBLISHED_PER_TOPIC, "5");
+		defaults.setProperty(ConfigKeys.NUMBER_OF_TOPICS_PUBLISHED, "5");
+		defaults.setProperty(ConfigKeys.IS_XPATH_ENABLED, "false");
+		defaults.setProperty(ConfigKeys.XPATH, "/c/b/a");
+		defaults.setProperty(ConfigKeys.PAYLOAD_MULTIPLYER, "1");
+		defaults.setProperty(ConfigKeys.PROTOCOL, "wse");
+		defaults.setProperty(ConfigKeys.PUBLISH_TIME_INTERVAL, "10000");
+		defaults.setProperty(ConfigKeys.PERFORMANCE_TEST_TIMEOUT, "5000000");
+		defaults.setProperty(ConfigKeys.NUMBER_OF_SUBSCRIBERS, "1");
+		defaults.setProperty(ConfigKeys.MULTI_THREAD_PER_SUB, "50");
+		return defaults;
+	}
+
+	private static void printStatistics(List<Stat> stats, File aFile)
+			throws IOException {
+		int maxLen = 0;
+		Writer output = new BufferedWriter(new FileWriter(aFile, true));
+
+		for (Stat stat : stats) {
+			maxLen = Math.max(maxLen, stat.name.length());
+		}
+
+		char[] fillchars = null;
+
+		for (Stat stat : stats) {
+			fillchars = new char[maxLen - stat.name.length() + 1];
+			Arrays.fill(fillchars, ' ');
+			String formattedStr = String.format("%s%s : %s", stat.name,
+					new String(fillchars), stat.value.toString());
+			output.write(formattedStr + "\n");
+			System.out.println(formattedStr);
+		}
+
+		fillchars = new char[maxLen];
+		Arrays.fill(fillchars, '-');
+		String fillingString = new String(fillchars);
+		output.write(fillingString + "\n");
+		System.out.println(fillingString);
+		output.close();
+	}
+
+	public static void loadConfigurationsFromFile() {
+		configurations = new Properties(getDefaults());
+
+		try {
+			URL url = ClassLoader
+					.getSystemResource(ConfigKeys.CONFIG_FILE_NAME);
+			if (url == null) {
+				throw new IOException("configuration file not found");
+			}
+			configurations.load(url.openStream());
+		} catch (IOException ioe) {
+			System.out
+					.println("unable to load configuration file, default settings will be used");
+		}
+	}
+
+	// Not used, If required to run as a test case call it from main
+	public static void loadConfigurationsFromSystemEnv() {
+
+		configurations = new Properties(getDefaults());
+
+		Properties envConfigs = System.getProperties();
+		String brokerUrl = envConfigs.getProperty(ConfigKeys.BROKER_URL, null);
+		String consumerUrl = envConfigs.getProperty(ConfigKeys.CONSUMER_EPR,
+				null);
+		String consumerPort = envConfigs.getProperty(ConfigKeys.CONSUMER_PORT,
+				null);
+		String isXpathEnabled = envConfigs.getProperty(
+				ConfigKeys.IS_XPATH_ENABLED, null);
+		String notifPerTopic = envConfigs.getProperty(
+				ConfigKeys.NOTIFICATIONS_PUBLISHED_PER_TOPIC, null);
+		String subsPerTopic = envConfigs.getProperty(
+				ConfigKeys.NUMBER_OF_SUBS_PERTOPIC, null);
+		String noTopicsPublished = envConfigs.getProperty(
+				ConfigKeys.NUMBER_OF_TOPICS_PUBLISHED, null);
+		String payLoadMultiplier = envConfigs.getProperty(
+				ConfigKeys.PAYLOAD_MULTIPLYER, null);
+		String protocol = envConfigs.getProperty(ConfigKeys.PROTOCOL, null);
+		String topicSimple = envConfigs.getProperty(ConfigKeys.TOPIC_SIMPLE,
+				null);
+		String topicXpath = envConfigs.getProperty(ConfigKeys.XPATH, null);
+
+		if (brokerUrl == null) {
+			System.err.println(ConfigKeys.BROKER_URL + " not given");
+			System.exit(1);
+		}
+		if (consumerUrl == null) {
+			System.err.println(ConfigKeys.CONSUMER_EPR + " not given");
+			System.exit(1);
+		}
+		if (consumerPort == null) {
+			System.err.println(ConfigKeys.CONSUMER_PORT + " not given");
+			System.exit(1);
+		}
+		if (isXpathEnabled == null) {
+			System.err.println(ConfigKeys.IS_XPATH_ENABLED + " not given");
+			System.exit(1);
+		}
+		if (notifPerTopic == null) {
+			System.err.println(ConfigKeys.NOTIFICATIONS_PUBLISHED_PER_TOPIC
+					+ " not given");
+			System.exit(1);
+		}
+		if (subsPerTopic == null) {
+			System.err.println(ConfigKeys.NUMBER_OF_SUBS_PERTOPIC
+					+ " not given");
+			System.exit(1);
+		}
+		if (noTopicsPublished == null) {
+			System.err.println(ConfigKeys.NUMBER_OF_TOPICS_PUBLISHED
+					+ " not given");
+			System.exit(1);
+		}
+		if (payLoadMultiplier == null) {
+			System.err.println(ConfigKeys.PAYLOAD_MULTIPLYER + " not given");
+			System.exit(1);
+		}
+		if (protocol == null) {
+			System.err.println(ConfigKeys.PROTOCOL + " not given");
+			System.exit(1);
+		}
+		if (topicSimple == null) {
+			System.err.println(ConfigKeys.TOPIC_SIMPLE + " not given");
+			System.exit(1);
+		}
+		if (topicXpath == null) {
+			System.err.println(ConfigKeys.XPATH + " not given");
+			System.exit(1);
+		}
+
+		configurations.put(ConfigKeys.BROKER_URL, brokerUrl);
+		configurations.put(ConfigKeys.CONSUMER_EPR, consumerUrl);
+		configurations.put(ConfigKeys.CONSUMER_PORT, consumerPort);
+		configurations.put(ConfigKeys.IS_XPATH_ENABLED, isXpathEnabled);
+		configurations.put(ConfigKeys.NOTIFICATIONS_PUBLISHED_PER_TOPIC,
+				notifPerTopic);
+		configurations.put(ConfigKeys.NUMBER_OF_SUBS_PERTOPIC, subsPerTopic);
+		configurations.put(ConfigKeys.NUMBER_OF_TOPICS_PUBLISHED,
+				noTopicsPublished);
+		configurations.put(ConfigKeys.PAYLOAD_MULTIPLYER, payLoadMultiplier);
+		configurations.put(ConfigKeys.PROTOCOL, protocol);
+		configurations.put(ConfigKeys.TOPIC_SIMPLE, topicSimple);
+		configurations.put(ConfigKeys.XPATH, topicXpath);
+	}
+}

Propchange: incubator/airavata/donations/ogce-donation/modules/ws-messaging/broker/src/test/java/performance_evaluator/rtt/PerformanceTest.java
------------------------------------------------------------------------------
    svn:executable = *

Added: incubator/airavata/donations/ogce-donation/modules/ws-messaging/broker/src/test/java/performance_evaluator/rtt/PublisherThread.java
URL: http://svn.apache.org/viewvc/incubator/airavata/donations/ogce-donation/modules/ws-messaging/broker/src/test/java/performance_evaluator/rtt/PublisherThread.java?rev=1142350&view=auto
==============================================================================
--- incubator/airavata/donations/ogce-donation/modules/ws-messaging/broker/src/test/java/performance_evaluator/rtt/PublisherThread.java (added)
+++ incubator/airavata/donations/ogce-donation/modules/ws-messaging/broker/src/test/java/performance_evaluator/rtt/PublisherThread.java Sun Jul  3 00:29:24 2011
@@ -0,0 +1,84 @@
+package performance_evaluator.rtt;
+
+import java.util.concurrent.CountDownLatch;
+
+import wsmg.client.WseClientAPI;
+import wsmg.client.WsmgClientAPI;
+import wsmg.client.WsntClientAPI;
+
+/**
+ * @author dinesh
+ */
+
+public class PublisherThread extends Thread {
+	private String brokerURL;
+	private String topic;
+	private final CountDownLatch startSignal;
+	private final CountDownLatch doneSignal;
+	private long totPublishTime =0l;
+	long avgPublishTime=0l;
+
+	private String payload = "";
+	String msg = "";
+	private WsmgClientAPI client = null;
+	int trackId = 0;
+	int threadId = 0;
+
+	public PublisherThread(String protocolIn, String brokerURLIn,
+			String topicIn, String payloadIn, CountDownLatch startSignalIn,
+			CountDownLatch doneSignalIn, int threadIdIn) {
+		this.payload = payloadIn;
+		this.brokerURL = brokerURLIn;
+		this.topic = topicIn;
+		this.startSignal = startSignalIn;
+		this.doneSignal = doneSignalIn;
+		this.threadId = threadIdIn;
+		if ("wse".equalsIgnoreCase(protocolIn)) {
+
+			WseClientAPI wseClient = new WseClientAPI();
+			wseClient.setTimeOutInMilliSeconds(0);
+			client = wseClient;
+
+		} else {
+
+			WsntClientAPI wsntClient = new WsntClientAPI();
+			wsntClient.setTimeOutInMilliSeconds(0);
+			client = wsntClient;
+		}
+
+	}
+
+	public void run() {
+
+		try {
+			trackId = 1;
+			startSignal.await();
+			System.out.println("Publishing started for topic :" + this.topic);
+			for (int i = 0; i < PerformanceTest.NOTIFICATIONS_PUBLISHED_PER_TOPIC; i++) {
+				msg = "<perf:performancetest xmlns:perf=\"http://lead.extreme.indiana.edu/namespaces/performance\"><perf:time>"
+						+ System.currentTimeMillis()
+						+ "</perf:time><perf:trackInfo><perf:threadId>"
+						+ threadId
+						+ "</perf:threadId><perf:trackId>"
+						+ trackId
+						+ "</perf:trackId></perf:trackInfo>"
+						+ "<perf:payload>"
+						+ payload + "</perf:payload></perf:performancetest>";
+				long publishStartTime = System.currentTimeMillis();
+				client.publish(brokerURL, topic, msg);
+				totPublishTime += System.currentTimeMillis()-publishStartTime;
+				trackId++;
+			}
+			
+			avgPublishTime=totPublishTime/PerformanceTest.NOTIFICATIONS_PUBLISHED_PER_TOPIC;
+			System.out.println("Publishing ended for topic :" + this.topic);
+			doneSignal.countDown();
+		} catch (Exception e) {
+			e.printStackTrace();
+		}
+	}
+	
+	synchronized long getAvgPubTime(){
+		return this.avgPublishTime;
+	}
+}

Propchange: incubator/airavata/donations/ogce-donation/modules/ws-messaging/broker/src/test/java/performance_evaluator/rtt/PublisherThread.java
------------------------------------------------------------------------------
    svn:executable = *

Added: incubator/airavata/donations/ogce-donation/modules/ws-messaging/broker/src/test/java/performance_evaluator/rtt/StatCalculatorThread.java
URL: http://svn.apache.org/viewvc/incubator/airavata/donations/ogce-donation/modules/ws-messaging/broker/src/test/java/performance_evaluator/rtt/StatCalculatorThread.java?rev=1142350&view=auto
==============================================================================
--- incubator/airavata/donations/ogce-donation/modules/ws-messaging/broker/src/test/java/performance_evaluator/rtt/StatCalculatorThread.java (added)
+++ incubator/airavata/donations/ogce-donation/modules/ws-messaging/broker/src/test/java/performance_evaluator/rtt/StatCalculatorThread.java Sun Jul  3 00:29:24 2011
@@ -0,0 +1,89 @@
+package performance_evaluator.rtt;
+
+import java.util.concurrent.TimeUnit;
+
+/**
+ * @author dinesh
+ */
+
+public class StatCalculatorThread extends Thread {
+
+	private NotificationManager notifManager = null;
+	private long lastMsgReceivedTime = 0l;
+	private long timeTot = 0;
+	private long avgTime = 0;
+	private int numberOfMssgsReceived = 0; // to avoid concurrency
+	private long timeOutMillis;
+	private int expectedNoMessages = 0;
+
+	public StatCalculatorThread(NotificationManager notificationManager,
+			long timeOutInMillis) throws Exception {
+		this.timeOutMillis = timeOutInMillis;
+		this.notifManager = notificationManager;
+		expectedNoMessages=PerformanceTest.NOTIFICATIONS_PUBLISHED_PER_TOPIC*notificationManager.getNoTopicsSubscribed();
+	}
+
+	@Override
+	public void run() {
+		do {
+
+			StatContainer container = null;
+			try {
+				container = notifManager.getQueue().poll(timeOutMillis,
+						TimeUnit.MILLISECONDS);
+
+				if (container != null) {
+					String trackInfo = "";
+					timeTot += container.getRondTripTime();
+					lastMsgReceivedTime = container.getMessageReceivedTime();
+					numberOfMssgsReceived++;
+					// ******un-comment in order to log trakId and message
+					// related other information*****
+					// if (logger.isInfoEnabled()) {
+					// trackInfo = env
+					// .getBody()
+					// .getFirstElement()
+					// .getFirstChildWithName(
+					// new QName(
+					// "http://lead.extreme.indiana.edu/namespaces/performance",
+					// "trackInfo")).toStringWithConsume();
+					// logger.info(trackInfo + "   Send time :" + time
+					// + "  Received time :" + System.currentTimeMillis());
+					// }
+				} else {
+					System.out
+							.println("stat calculator thread was interrupted");
+					break;
+				}
+			} catch (InterruptedException e1) {
+				e1.printStackTrace();
+				break;
+			}
+
+		} while (expectedNoMessages > numberOfMssgsReceived);
+
+		if (numberOfMssgsReceived > 0) {
+			avgTime = timeTot / numberOfMssgsReceived;
+		} else {
+			System.out.println("no messages received");
+		}
+
+		System.out.println("end of stat calculator");
+	}
+
+	synchronized long getTotalTime() {
+		return timeTot;
+	}
+
+	synchronized long getAverageTime() {
+		return avgTime;
+	}
+
+	synchronized long getNumberOfMsgReceived() {
+		return numberOfMssgsReceived;
+	}
+
+	synchronized public long getLastMsgReceivedTime() {
+		return lastMsgReceivedTime;
+	}
+}

Propchange: incubator/airavata/donations/ogce-donation/modules/ws-messaging/broker/src/test/java/performance_evaluator/rtt/StatCalculatorThread.java
------------------------------------------------------------------------------
    svn:executable = *

Added: incubator/airavata/donations/ogce-donation/modules/ws-messaging/broker/src/test/java/performance_evaluator/rtt/StatContainer.java
URL: http://svn.apache.org/viewvc/incubator/airavata/donations/ogce-donation/modules/ws-messaging/broker/src/test/java/performance_evaluator/rtt/StatContainer.java?rev=1142350&view=auto
==============================================================================
--- incubator/airavata/donations/ogce-donation/modules/ws-messaging/broker/src/test/java/performance_evaluator/rtt/StatContainer.java (added)
+++ incubator/airavata/donations/ogce-donation/modules/ws-messaging/broker/src/test/java/performance_evaluator/rtt/StatContainer.java Sun Jul  3 00:29:24 2011
@@ -0,0 +1,29 @@
+package performance_evaluator.rtt;
+
+import org.apache.axiom.soap.SOAPEnvelope;
+
+public class StatContainer {
+	private SOAPEnvelope msgEnvelope =null;
+	private long rtt=0l;
+	private long receivedTime = 0l;
+	
+	
+	public StatContainer(SOAPEnvelope msgEnvelope){
+		this.msgEnvelope = msgEnvelope;
+		this.receivedTime  = System.currentTimeMillis();
+		this.rtt = this.receivedTime -Long.parseLong(msgEnvelope.getBody().getFirstElement()
+				.getFirstElement().getText());
+	}
+
+	public long getRondTripTime(){
+		return this.rtt;
+	}
+	
+	public SOAPEnvelope getMsgEnvelope(){
+		return msgEnvelope;
+	}
+	
+	public long getMessageReceivedTime(){
+		return this.receivedTime;
+	}
+}

Added: incubator/airavata/donations/ogce-donation/modules/ws-messaging/broker/src/test/java/performance_evaluator/rtt/Subscription.java
URL: http://svn.apache.org/viewvc/incubator/airavata/donations/ogce-donation/modules/ws-messaging/broker/src/test/java/performance_evaluator/rtt/Subscription.java?rev=1142350&view=auto
==============================================================================
--- incubator/airavata/donations/ogce-donation/modules/ws-messaging/broker/src/test/java/performance_evaluator/rtt/Subscription.java (added)
+++ incubator/airavata/donations/ogce-donation/modules/ws-messaging/broker/src/test/java/performance_evaluator/rtt/Subscription.java Sun Jul  3 00:29:24 2011
@@ -0,0 +1,88 @@
+package performance_evaluator.rtt;
+
+import org.apache.axis2.addressing.EndpointReference;
+import org.apache.log4j.Logger;
+
+import wsmg.client.ConsumerNotificationHandler;
+import wsmg.client.WsmgClientAPI;
+
+
+public class Subscription {
+
+	private String subscriptionID;
+
+	private String topic;
+	private final static org.apache.log4j.Logger logger = Logger
+			.getLogger(Subscription.class);
+	private ConsumerNotificationHandler handler;
+	private WsmgClientAPI client;
+	private EndpointReference messageBoxEPR;
+	private String xpath;
+	private String brokerURL;
+
+	private String protocol;
+
+	public Subscription(WsmgClientAPI clientIn, String subscriptionID, String topic,
+			ConsumerNotificationHandler callback, String brokerURL, String protocolIn) {
+		super();
+		this.subscriptionID = subscriptionID;
+		this.topic = topic;
+		this.handler = callback;
+		this.brokerURL = brokerURL;
+		this.client = clientIn;
+		this.protocol = protocolIn;
+	}
+	
+	public Subscription(WsmgClientAPI clientIn, String subscriptionID, String topic,String xpath,
+			ConsumerNotificationHandler callback, String brokerURL, String protocolIn) {
+		super();
+		this.client = clientIn;
+		this.subscriptionID = subscriptionID;
+		this.topic = topic;
+		this.handler = callback;
+		this.brokerURL = brokerURL;
+		this.xpath=xpath;
+		this.protocol = protocolIn;
+	}
+
+	public ConsumerNotificationHandler getCallback() {
+		return handler;
+	}
+
+	public String getTopic() {
+		return topic;
+	}
+
+//	public void destroy() throws RemoteException {
+//		client.shutdownConsumerService();
+//	}
+
+	public EndpointReference getMessageBoxEPR() {
+		return messageBoxEPR;
+	}
+
+	public void setMessageBoxEpr(EndpointReference messageBoxEPR) {
+		this.messageBoxEPR = messageBoxEPR;
+	}
+
+	public String getSubscriptionID() {
+		return subscriptionID;
+	}
+
+	public void setSubscriptionID(String subscriptionID) {
+		this.subscriptionID = subscriptionID;
+	}
+
+	public String getBrokerURL() {
+		return brokerURL;
+	}
+
+	public void setBrokerURL(String brokerURL) {
+		this.brokerURL = brokerURL;
+	}
+
+//	public String getConsumerEPR() throws UnknownHostException {
+//		cli
+//	}
+
+}

Propchange: incubator/airavata/donations/ogce-donation/modules/ws-messaging/broker/src/test/java/performance_evaluator/rtt/Subscription.java
------------------------------------------------------------------------------
    svn:executable = *

Added: incubator/airavata/donations/ogce-donation/modules/ws-messaging/broker/src/test/java/performance_evaluator/rtt/util/ConfigKeys.java
URL: http://svn.apache.org/viewvc/incubator/airavata/donations/ogce-donation/modules/ws-messaging/broker/src/test/java/performance_evaluator/rtt/util/ConfigKeys.java?rev=1142350&view=auto
==============================================================================
--- incubator/airavata/donations/ogce-donation/modules/ws-messaging/broker/src/test/java/performance_evaluator/rtt/util/ConfigKeys.java (added)
+++ incubator/airavata/donations/ogce-donation/modules/ws-messaging/broker/src/test/java/performance_evaluator/rtt/util/ConfigKeys.java Sun Jul  3 00:29:24 2011
@@ -0,0 +1,26 @@
+package performance_evaluator.rtt.util;
+
+/**
+ * @author dinesh
+ */
+
+public interface ConfigKeys {
+	String CONFIG_FILE_NAME = "configurations.properties";
+	String BROKER_URL = "broker.eventing.service.epr";
+	String CONSUMER_EPR = "consumer.location";
+	String CONSUMER_PORT = "consumer.port";
+	String TOPIC_SIMPLE = "topic.simple";
+	String TOPIC_XPATH = "topic.xpath";
+	String PUBLISH_TIME_INTERVAL = "publish.time.interval";
+	String IS_XPATH_ENABLED = "is.xpath.enabled";
+	String XPATH = "topic.xpath";
+	String PAYLOAD_MULTIPLYER = "payload.multiplyer";
+	String PROTOCOL = "protocol.used";
+	String NUMBER_OF_SUBS_PERTOPIC = "num.subscribers.per.topic";
+	String NOTIFICATIONS_PUBLISHED_PER_TOPIC = "notifications.per.topic";
+	String NUMBER_OF_TOPICS_PUBLISHED = "number.of.topics";
+	String SCHEDULER_REPEAT_PERIOD = "stat.timeout.monitor.scheduler.period";
+	String PERFORMANCE_TEST_TIMEOUT = "performance.test.timeout.period.millis";
+	String NUMBER_OF_SUBSCRIBERS = "number.of.subscriber.servers";
+	String MULTI_THREAD_PER_SUB="num.muti.thread.per.sub";
+}

Propchange: incubator/airavata/donations/ogce-donation/modules/ws-messaging/broker/src/test/java/performance_evaluator/rtt/util/ConfigKeys.java
------------------------------------------------------------------------------
    svn:executable = *

Added: incubator/airavata/donations/ogce-donation/modules/ws-messaging/broker/src/test/java/performance_evaluator/rtt/util/LoadMsgPayload.java
URL: http://svn.apache.org/viewvc/incubator/airavata/donations/ogce-donation/modules/ws-messaging/broker/src/test/java/performance_evaluator/rtt/util/LoadMsgPayload.java?rev=1142350&view=auto
==============================================================================
--- incubator/airavata/donations/ogce-donation/modules/ws-messaging/broker/src/test/java/performance_evaluator/rtt/util/LoadMsgPayload.java (added)
+++ incubator/airavata/donations/ogce-donation/modules/ws-messaging/broker/src/test/java/performance_evaluator/rtt/util/LoadMsgPayload.java Sun Jul  3 00:29:24 2011
@@ -0,0 +1,45 @@
+package performance_evaluator.rtt.util;
+
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.InputStreamReader;
+import java.net.URL;
+
+public class LoadMsgPayload {
+
+	private static LoadMsgPayload msg = null;
+
+	public static LoadMsgPayload getInstance() {
+		if (msg == null)
+			msg = new LoadMsgPayload();
+		return msg;
+	}
+
+	public String getMessage(String fileName) throws IOException {
+		URL url = ClassLoader.getSystemResource(fileName);
+		if(url!=null)
+			return convertStreamToString(url.openStream());
+		else return "";
+	}
+
+	private String convertStreamToString(InputStream is) throws IOException {
+		if (is != null) {
+			StringBuilder sb = new StringBuilder();
+			String line;
+
+			try {
+				BufferedReader reader = new BufferedReader(
+						new InputStreamReader(is, "UTF-8"));
+				while ((line = reader.readLine()) != null) {
+					sb.append(line).append("\n");
+				}
+			} finally {
+				is.close();
+			}
+			return sb.toString();
+		} else {
+			return "";
+		}
+	}
+}

Propchange: incubator/airavata/donations/ogce-donation/modules/ws-messaging/broker/src/test/java/performance_evaluator/rtt/util/LoadMsgPayload.java
------------------------------------------------------------------------------
    svn:executable = *

Added: incubator/airavata/donations/ogce-donation/modules/ws-messaging/broker/src/test/java/performance_evaluator/rtt/util/LoadXpath.java
URL: http://svn.apache.org/viewvc/incubator/airavata/donations/ogce-donation/modules/ws-messaging/broker/src/test/java/performance_evaluator/rtt/util/LoadXpath.java?rev=1142350&view=auto
==============================================================================
--- incubator/airavata/donations/ogce-donation/modules/ws-messaging/broker/src/test/java/performance_evaluator/rtt/util/LoadXpath.java (added)
+++ incubator/airavata/donations/ogce-donation/modules/ws-messaging/broker/src/test/java/performance_evaluator/rtt/util/LoadXpath.java Sun Jul  3 00:29:24 2011
@@ -0,0 +1,51 @@
+package performance_evaluator.rtt.util;
+
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.InputStreamReader;
+import java.net.URL;
+import java.util.LinkedList;
+
+/**
+ * @author dinesh
+ */
+
+public class LoadXpath {
+	private static LoadXpath xpath = null;
+	LinkedList<String> xpathList = null;
+
+	public static LoadXpath getInstace() {
+		if (xpath == null)
+			xpath = new LoadXpath();
+		return xpath;
+	}
+
+	public LinkedList<String> getXpathList(String fileName) throws IOException {
+		URL url = ClassLoader.getSystemResource(fileName);
+		if (url != null && xpathList == null)
+			return convertStreamToString(url.openStream());
+		return xpathList;
+	}
+
+	private LinkedList<String> convertStreamToString(InputStream is)
+			throws IOException {
+		if (is != null) {
+			xpathList = new LinkedList<String>();
+			String line;
+
+			try {
+				BufferedReader reader = new BufferedReader(
+						new InputStreamReader(is, "UTF-8"));
+				while ((line = reader.readLine()) != null) {
+					xpathList.add(line);
+				}
+			} finally {
+				is.close();
+			}
+			return xpathList;
+		} else {
+			return null;
+		}
+	}
+}

Added: incubator/airavata/donations/ogce-donation/modules/ws-messaging/broker/src/test/java/wsmg/broker/BrokerWSETest.java
URL: http://svn.apache.org/viewvc/incubator/airavata/donations/ogce-donation/modules/ws-messaging/broker/src/test/java/wsmg/broker/BrokerWSETest.java?rev=1142350&view=auto
==============================================================================
--- incubator/airavata/donations/ogce-donation/modules/ws-messaging/broker/src/test/java/wsmg/broker/BrokerWSETest.java (added)
+++ incubator/airavata/donations/ogce-donation/modules/ws-messaging/broker/src/test/java/wsmg/broker/BrokerWSETest.java Sun Jul  3 00:29:24 2011
@@ -0,0 +1,100 @@
+package wsmg.broker;
+
+import java.io.IOException;
+import java.util.Properties;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+
+import junit.framework.TestCase;
+
+import org.apache.axiom.soap.SOAPEnvelope;
+import org.apache.axis2.AxisFault;
+import org.junit.Test;
+
+import wsmg.client.ConsumerNotificationHandler;
+import wsmg.client.WseClientAPI;
+
+public class BrokerWSETest extends TestCase implements
+		ConsumerNotificationHandler {
+
+	static Properties configs = new Properties();
+
+	BlockingQueue<SOAPEnvelope> queue = new LinkedBlockingQueue<SOAPEnvelope>();
+
+	public void handleNotification(SOAPEnvelope msgEnvelope) {
+		queue.add(msgEnvelope);
+		System.out.println("Received " + msgEnvelope);
+	}
+
+	BlockingQueue<SOAPEnvelope> getMsgQueue() {
+		return queue;
+	}
+
+	@Override
+	protected void setUp() throws Exception {
+		//TestUtilServer.start(null, null);
+	}
+
+	@Test
+	public void testRoundTrip() throws InterruptedException {
+		
+		try {
+
+			String brokerEPR = "http://127.0.0.1:8080/axis2/services/EventingService";
+			long value = System.currentTimeMillis();
+			String msg = String.format("<msg> current time is : %d </msg>",
+					value);
+
+			WseClientAPI clientApi = new WseClientAPI(null, 1000000);
+
+			int consumerPort = 6767;
+
+			String[] consumerEPRs = clientApi.startConsumerService(
+					consumerPort, this);
+
+			assertTrue(consumerEPRs.length > 0);
+
+			String topic = "WseRoundTripTestTopic";
+
+			String subscriptionID = clientApi.subscribe(brokerEPR,
+					consumerEPRs[0], topic);
+			String subscriptionID2 = clientApi.subscribe(brokerEPR,
+					consumerEPRs[0], topic, "/foo/bar");
+
+			clientApi.publish(brokerEPR, topic, msg);
+
+			clientApi.publish(brokerEPR, topic, "<foo><bar>Test</bar></foo>");
+
+			int receivedMessageCount = 0;
+
+			Thread.sleep(2000);
+
+			try {
+				clientApi.unSubscribe(brokerEPR, subscriptionID, null);
+			} catch (AxisFault e) {
+				// TODO Auto-generated catch block
+				e.printStackTrace();
+			}
+			clientApi.shutdownConsumerService();
+
+		} catch (AxisFault e) {
+			e.printStackTrace();
+			try {
+				System.in.read();
+			} catch (IOException e1) {
+				// TODO Auto-generated catch block
+				e1.printStackTrace();
+			}
+
+			// fail("unexpected exception occured");
+		}
+		System.out.println("Broker roundtrip done");
+
+	}
+
+	@Override
+	protected void tearDown() throws Exception {
+		//TestUtilServer.stop();
+	}
+
+}

Added: incubator/airavata/donations/ogce-donation/modules/ws-messaging/broker/src/test/java/wsmg/broker/BrokerWSNTTest.java
URL: http://svn.apache.org/viewvc/incubator/airavata/donations/ogce-donation/modules/ws-messaging/broker/src/test/java/wsmg/broker/BrokerWSNTTest.java?rev=1142350&view=auto
==============================================================================
--- incubator/airavata/donations/ogce-donation/modules/ws-messaging/broker/src/test/java/wsmg/broker/BrokerWSNTTest.java (added)
+++ incubator/airavata/donations/ogce-donation/modules/ws-messaging/broker/src/test/java/wsmg/broker/BrokerWSNTTest.java Sun Jul  3 00:29:24 2011
@@ -0,0 +1,105 @@
+package wsmg.broker;
+
+import java.io.IOException;
+import java.util.Properties;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+
+import junit.framework.TestCase;
+
+import org.apache.axiom.soap.SOAPEnvelope;
+import org.apache.axis2.AxisFault;
+import org.junit.Test;
+
+import wsmg.client.ConsumerNotificationHandler;
+import wsmg.client.WsntClientAPI;
+import wsmg.util.test.TestUtilServer;
+
+public class BrokerWSNTTest extends TestCase implements
+		ConsumerNotificationHandler {
+
+	static Properties configs = new Properties();
+
+	BlockingQueue<SOAPEnvelope> queue = new LinkedBlockingQueue<SOAPEnvelope>();
+
+	public void handleNotification(SOAPEnvelope msgEnvelope) {
+		// queue.add(msgEnvelope);
+		System.out.println("Received " + msgEnvelope);
+	}
+
+	BlockingQueue<SOAPEnvelope> getMsgQueue() {
+		return queue;
+	}
+
+	@Override
+	protected void setUp() throws Exception {
+		TestUtilServer.start(null, null);
+	}
+
+	@Test
+	public void testRoundTrip() throws InterruptedException {
+
+		try {
+			long value = System.currentTimeMillis();
+			String msg = String.format("<msg> current time is : %d </msg>",
+					value);
+
+			WsntClientAPI clientApi = new WsntClientAPI(1000000);
+
+			int consumerPort = 6767;
+
+			String brokerEPR = "http://127.0.0.1:5555/axis2/services/NotificationService";
+			String[] consumerEPRs = clientApi.startConsumerService(
+					consumerPort, this);
+
+			assertTrue(consumerEPRs.length > 0);
+
+			String topic = "WsntRoundTripTestTopic";
+
+			String topicSubscriptionID = clientApi.subscribe(brokerEPR,
+					consumerEPRs[0], topic);
+
+			System.out.println("topic subscription id: " + topicSubscriptionID);
+
+			String xpathSubscriptionID = clientApi.subscribe(brokerEPR,
+					consumerEPRs[0], topic, "/foo/bar");
+
+			System.out.println("xpath subscription id: " + xpathSubscriptionID);
+
+			clientApi.publish(brokerEPR, topic, msg);
+
+			clientApi.publish(brokerEPR, topic, "<foo><bar>eligible to</bar></foo>");
+
+			Thread.sleep(2000);
+
+			try {
+				clientApi.unSubscribe(brokerEPR, topicSubscriptionID, null);
+				clientApi.unSubscribe(brokerEPR, xpathSubscriptionID, null);
+			} catch (AxisFault e) {
+				// TODO Auto-generated catch block
+				e.printStackTrace();
+			}
+			
+			clientApi.shutdownConsumerService();
+
+		} catch (AxisFault e) {
+			e.printStackTrace();
+			try {
+				System.in.read();
+			} catch (IOException e1) {
+				// TODO Auto-generated catch block
+				e1.printStackTrace();
+			}
+
+			// fail("unexpected exception occured");
+		}
+		System.out.println("Broker roundtrip done");
+
+	}
+
+	@Override
+	protected void tearDown() throws Exception {
+		TestUtilServer.stop();
+	}
+
+}

Added: incubator/airavata/donations/ogce-donation/modules/ws-messaging/broker/src/test/java/wsmg/client/TestWseMsgBrokerClient.java
URL: http://svn.apache.org/viewvc/incubator/airavata/donations/ogce-donation/modules/ws-messaging/broker/src/test/java/wsmg/client/TestWseMsgBrokerClient.java?rev=1142350&view=auto
==============================================================================
--- incubator/airavata/donations/ogce-donation/modules/ws-messaging/broker/src/test/java/wsmg/client/TestWseMsgBrokerClient.java (added)
+++ incubator/airavata/donations/ogce-donation/modules/ws-messaging/broker/src/test/java/wsmg/client/TestWseMsgBrokerClient.java Sun Jul  3 00:29:24 2011
@@ -0,0 +1,100 @@
+package wsmg.client;
+
+import static org.junit.Assert.assertTrue;
+
+import java.io.IOException;
+import java.util.concurrent.LinkedBlockingQueue;
+
+import org.apache.axiom.soap.SOAPEnvelope;
+import org.apache.axis2.AxisFault;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+import wsmg.util.test.TestUtilServer;
+
+public class TestWseMsgBrokerClient {
+
+	@Before
+	public void setUp() throws Exception {
+	}
+
+	@After
+	public void tearDown() throws Exception {
+	}
+
+	@Test
+	public void testRoundTrip() throws Exception {
+
+		TestUtilServer.start(null, null);
+
+		final LinkedBlockingQueue<SOAPEnvelope> queue = new LinkedBlockingQueue<SOAPEnvelope>();
+
+		ConsumerNotificationHandler handler = new ConsumerNotificationHandler() {
+
+			public void handleNotification(SOAPEnvelope msgEnvelope)
+					throws AxisFault {
+				queue.add(msgEnvelope);
+				System.out.println(msgEnvelope.toString());
+			}
+		};
+
+		try {
+
+			String brokerEPR = "http://127.0.0.1:"
+					+ TestUtilServer.TESTING_PORT
+					+ "/axis2/services/NotificationService";
+			long value = System.currentTimeMillis();
+			String msg = String.format("<msg> current time is : %d </msg>",
+					value);
+
+			WsntMsgBrokerClient client = new WsntMsgBrokerClient();
+			client.init(brokerEPR);
+
+			int consumerPort = 6767;
+
+			String[] consumerEPRs = client.startConsumerService(consumerPort,
+					handler);
+
+			assertTrue(consumerEPRs.length > 0);
+
+			String topic = "WsntRoundTripTestTopic";
+
+			String subscriptionID = client.subscribe(consumerEPRs[0], topic,
+					null);
+
+			System.out.println("topic sub id = " + subscriptionID);
+
+			String subscriptionID2 = client.subscribe(consumerEPRs[0], topic,
+					"/foo/bar");
+
+			System.out.println("xpath sub id = " + subscriptionID2);
+
+			client.publish(topic, msg);
+
+			client.publish(topic, "<foo><bar>Test</bar></foo>");
+
+			Thread.sleep(2000);
+
+			client.unSubscribe(subscriptionID);
+
+			client.shutdownConsumerService();
+
+		} catch (MsgBrokerClientException e) {
+			e.printStackTrace();
+			try {
+				System.in.read();
+			} catch (IOException e1) {
+				// TODO Auto-generated catch block
+				e1.printStackTrace();
+			}
+
+			// fail("unexpected exception occured");
+		}
+
+		TestUtilServer.stop();
+		System.out.println("Broker roundtrip done");
+
+	}
+
+}

Added: incubator/airavata/donations/ogce-donation/modules/ws-messaging/broker/src/test/java/wsmg/client/TestWsntMsgBrokerClient.java
URL: http://svn.apache.org/viewvc/incubator/airavata/donations/ogce-donation/modules/ws-messaging/broker/src/test/java/wsmg/client/TestWsntMsgBrokerClient.java?rev=1142350&view=auto
==============================================================================
--- incubator/airavata/donations/ogce-donation/modules/ws-messaging/broker/src/test/java/wsmg/client/TestWsntMsgBrokerClient.java (added)
+++ incubator/airavata/donations/ogce-donation/modules/ws-messaging/broker/src/test/java/wsmg/client/TestWsntMsgBrokerClient.java Sun Jul  3 00:29:24 2011
@@ -0,0 +1,100 @@
+package wsmg.client;
+
+import static org.junit.Assert.assertTrue;
+
+import java.io.IOException;
+import java.util.concurrent.LinkedBlockingQueue;
+
+import org.apache.axiom.soap.SOAPEnvelope;
+import org.apache.axis2.AxisFault;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+import wsmg.util.test.TestUtilServer;
+
+public class TestWsntMsgBrokerClient {
+
+	@Before
+	public void setUp() throws Exception {
+	}
+
+	@After
+	public void tearDown() throws Exception {
+	}
+
+	@Test
+	public void testRoundTrip() throws Exception {
+
+		TestUtilServer.start(null, null);
+
+		final LinkedBlockingQueue<SOAPEnvelope> queue = new LinkedBlockingQueue<SOAPEnvelope>();
+
+		ConsumerNotificationHandler handler = new ConsumerNotificationHandler() {
+
+			public void handleNotification(SOAPEnvelope msgEnvelope)
+					throws AxisFault {
+				queue.add(msgEnvelope);
+				System.out.println(msgEnvelope.toString());
+			}
+		};
+
+		try {
+
+			String brokerEPR = "http://127.0.0.1:"
+					+ TestUtilServer.TESTING_PORT
+					+ "/axis2/services/EventingService";
+			long value = System.currentTimeMillis();
+			String msg = String.format("<msg> current time is : %d </msg>",
+					value);
+
+			WseMsgBrokerClient client = new WseMsgBrokerClient();
+			client.init(brokerEPR);
+
+			int consumerPort = 6767;
+
+			String[] consumerEPRs = client.startConsumerService(consumerPort,
+					handler);
+
+			assertTrue(consumerEPRs.length > 0);
+
+			String topic = "WseRoundTripTestTopic";
+
+			String subscriptionID = client.subscribe(consumerEPRs[0], topic,
+					null);
+
+			System.out.println("topic sub id = " + subscriptionID);
+
+			String subscriptionID2 = client.subscribe(consumerEPRs[0], topic,
+					"/foo/bar");
+
+			System.out.println("xpath sub id = " + subscriptionID2);
+
+			client.publish(topic, msg);
+
+			client.publish(topic, "<foo><bar>Test</bar></foo>");
+
+			Thread.sleep(2000);
+
+			client.unSubscribe(subscriptionID);
+
+			client.shutdownConsumerService();
+
+		} catch (MsgBrokerClientException e) {
+			e.printStackTrace();
+			try {
+				System.in.read();
+			} catch (IOException e1) {
+				// TODO Auto-generated catch block
+				e1.printStackTrace();
+			}
+
+			// fail("unexpected exception occured");
+		}
+
+		TestUtilServer.stop();
+		System.out.println("Broker roundtrip done");
+
+	}
+
+}

Added: incubator/airavata/donations/ogce-donation/modules/ws-messaging/broker/src/test/java/wsmg/client/WseClientAPIConsumerTest.java
URL: http://svn.apache.org/viewvc/incubator/airavata/donations/ogce-donation/modules/ws-messaging/broker/src/test/java/wsmg/client/WseClientAPIConsumerTest.java?rev=1142350&view=auto
==============================================================================
--- incubator/airavata/donations/ogce-donation/modules/ws-messaging/broker/src/test/java/wsmg/client/WseClientAPIConsumerTest.java (added)
+++ incubator/airavata/donations/ogce-donation/modules/ws-messaging/broker/src/test/java/wsmg/client/WseClientAPIConsumerTest.java Sun Jul  3 00:29:24 2011
@@ -0,0 +1,100 @@
+/*
+ * Licensed to the Lanka Software Foundation (LSF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The LSF 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 wsmg.client;
+
+import java.net.URL;
+import java.util.Properties;
+
+import javax.xml.stream.XMLStreamException;
+
+import junit.framework.TestCase;
+
+import org.apache.axiom.soap.SOAPEnvelope;
+import org.apache.axis2.AxisFault;
+import org.junit.After;
+import org.junit.AfterClass;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import wsmg.util.ConfigKeys;
+
+
+
+public class WseClientAPIConsumerTest extends TestCase {
+
+	static Properties configs = new Properties();
+
+	@BeforeClass
+	public static void setUpBeforeClass() throws Exception {
+
+	}
+
+	@AfterClass
+	public static void tearDownAfterClass() throws Exception {
+	}
+
+	@Before
+	public void setUp() throws Exception {
+		URL configURL = ClassLoader
+				.getSystemResource(ConfigKeys.CONFIG_FILE_NAME);
+		configs.load(configURL.openStream());
+
+	}
+
+	@After
+	public void tearDown() throws Exception {
+	}
+
+	// basic subscription test
+	@Test
+	public final void testConsumerService() {
+
+		ConsumerNotificationHandler handler = new ConsumerNotificationHandler() {
+
+			public void handleNotification(SOAPEnvelope msgEnvelope) {
+				try {
+					System.out.println(msgEnvelope.toStringWithConsume());
+				} catch (XMLStreamException e) {
+					// TODO Auto-generated catch block
+					e.printStackTrace();
+				}
+			}
+		};
+
+		try {
+			WseClientAPI clientApi = new WseClientAPI();
+
+			int consumerPort = new Integer(configs
+					.getProperty(ConfigKeys.CONSUMER_PORT));
+
+			clientApi.startConsumerService(consumerPort, handler);
+			clientApi.shutdownConsumerService();
+			
+		} catch (AxisFault e) {
+			e.printStackTrace();
+			fail("unable to start consumer service: " + e.toString());
+		}
+
+	}
+
+}

Added: incubator/airavata/donations/ogce-donation/modules/ws-messaging/broker/src/test/java/wsmg/client/WseClientAPIPublishTest.java
URL: http://svn.apache.org/viewvc/incubator/airavata/donations/ogce-donation/modules/ws-messaging/broker/src/test/java/wsmg/client/WseClientAPIPublishTest.java?rev=1142350&view=auto
==============================================================================
--- incubator/airavata/donations/ogce-donation/modules/ws-messaging/broker/src/test/java/wsmg/client/WseClientAPIPublishTest.java (added)
+++ incubator/airavata/donations/ogce-donation/modules/ws-messaging/broker/src/test/java/wsmg/client/WseClientAPIPublishTest.java Sun Jul  3 00:29:24 2011
@@ -0,0 +1,129 @@
+/*
+ * Licensed to the Lanka Software Foundation (LSF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The LSF 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 wsmg.client;
+
+import java.net.URL;
+import java.util.Properties;
+
+import junit.framework.TestCase;
+
+import org.apache.axis2.AxisFault;
+import org.apache.axis2.addressing.EndpointReference;
+import org.junit.After;
+import org.junit.AfterClass;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import wsmg.util.ConfigKeys;
+
+public class WseClientAPIPublishTest extends TestCase {
+
+	static Properties configs = new Properties();
+
+	@BeforeClass
+	public static void setUpBeforeClass() throws Exception {
+
+	}
+
+	@AfterClass
+	public static void tearDownAfterClass() throws Exception {
+	}
+
+	@Before
+	public void setUp() throws Exception {
+		URL configURL = ClassLoader
+				.getSystemResource(ConfigKeys.CONFIG_FILE_NAME);
+		configs.load(configURL.openStream());
+
+	}
+
+	@After
+	public void tearDown() throws Exception {
+	}
+
+	// basic pubilsh test
+	@Test
+	public final void testPublishWithBrokerURL() {
+
+		try {
+			WseClientAPI clientApi = new WseClientAPI();
+
+			String brokerEPR = configs
+					.getProperty(ConfigKeys.BROKER_EVENTING_SERVICE_EPR);
+			String topic = "MYSimpleTOPIC";
+
+			String message = "<msg> test message </msg>";
+			// should throw an exception
+			clientApi.publish(brokerEPR, topic, message);
+
+		} catch (AxisFault e) {
+			e.printStackTrace();
+			fail("unable to publish: " + e.toString());
+		}
+
+	}
+
+	// basic publish test
+	@Test(expected = AxisFault.class)
+	public final void testPublishInvalidBrokerLocation() {
+
+		WseClientAPI clientApi = new WseClientAPI();
+
+		String brokerEPR = "http://invalid/url/";
+		String topic = "MYSimpleTOPIC";
+
+		String message = "<msg> test message </msg>";
+		// should throw an exception
+		try {
+			clientApi.publish(brokerEPR, topic, message);
+			fail("didn't throw an exception");
+		} catch (AxisFault e) {
+			// TODO Auto-generated catch block
+			e.printStackTrace();
+		}
+
+	}
+
+	// basic pubilsh test
+	@Test
+	public final void testPublishPlainText() {
+
+		try {
+			String brokerURL = configs
+					.getProperty(ConfigKeys.BROKER_EVENTING_SERVICE_EPR);
+			String topic = "MYSimpleTOPIC";
+
+			// should throw an exception
+			EndpointReference brokerEPR = new EndpointReference(brokerURL);
+			WseClientAPI clientApi = new WseClientAPI(brokerEPR);
+
+			clientApi.publishPlainText("Plain text to publish with out topic");
+			clientApi.publishPlainText(brokerURL, topic,
+					"Plain text to publish with topic");
+
+		} catch (AxisFault e) {
+			e.printStackTrace();
+			fail("unable to publish: " + e.toString());
+		}
+
+	}
+
+}

Added: incubator/airavata/donations/ogce-donation/modules/ws-messaging/broker/src/test/java/wsmg/client/WseClientAPIRoundTripTests.java
URL: http://svn.apache.org/viewvc/incubator/airavata/donations/ogce-donation/modules/ws-messaging/broker/src/test/java/wsmg/client/WseClientAPIRoundTripTests.java?rev=1142350&view=auto
==============================================================================
--- incubator/airavata/donations/ogce-donation/modules/ws-messaging/broker/src/test/java/wsmg/client/WseClientAPIRoundTripTests.java (added)
+++ incubator/airavata/donations/ogce-donation/modules/ws-messaging/broker/src/test/java/wsmg/client/WseClientAPIRoundTripTests.java Sun Jul  3 00:29:24 2011
@@ -0,0 +1,229 @@
+/*
+ * Licensed to the Lanka Software Foundation (LSF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The LSF 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 wsmg.client;
+
+import java.net.URL;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Properties;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+
+import junit.framework.TestCase;
+
+import org.apache.axiom.om.OMElement;
+import org.apache.axiom.soap.SOAPEnvelope;
+import org.apache.axis2.AxisFault;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+import wsmg.util.ConfigKeys;
+
+/**
+ * @author ramith
+ * 
+ */
+public class WseClientAPIRoundTripTests extends TestCase implements
+		ConsumerNotificationHandler {
+
+	static Properties configs = new Properties();
+
+	BlockingQueue<SOAPEnvelope> queue = new LinkedBlockingQueue<SOAPEnvelope>();
+
+	public void handleNotification(SOAPEnvelope msgEnvelope) {
+
+		queue.add(msgEnvelope);
+		System.out.println(msgEnvelope);
+	}
+
+	BlockingQueue<SOAPEnvelope> getMsgQueue() {
+		return queue;
+	}
+
+	/**
+	 * @throws java.lang.Exception
+	 */
+	@Before
+	public void setUp() throws Exception {
+		URL configURL = ClassLoader
+				.getSystemResource(ConfigKeys.CONFIG_FILE_NAME);
+		configs.load(configURL.openStream());
+	}
+
+	/**
+	 * @throws java.lang.Exception
+	 */
+	@After
+	public void tearDown() throws Exception {
+	}
+
+	@Test
+	public final void testRoundTrip() {
+
+		try {
+
+			long value = System.currentTimeMillis();
+			String msg = String.format("<msg> current time is : %d </msg>",
+					value);
+
+			WseClientAPI clientApi = new WseClientAPI();
+
+			int consumerPort = new Integer(configs
+					.getProperty(ConfigKeys.CONSUMER_PORT));
+
+			String brokerEPR = configs
+					.getProperty(ConfigKeys.BROKER_EVENTING_SERVICE_EPR);
+			String[] consumerEPRs = clientApi.startConsumerService(
+					consumerPort, this);
+
+			assertTrue(consumerEPRs.length > 0);
+
+			String topic = "WseRoundTripTestTopic";
+
+			String subscriptionID = clientApi.subscribe(brokerEPR,
+					consumerEPRs[0], topic);
+
+			clientApi.publish(brokerEPR, topic, msg);
+
+			while (true) {
+				try {
+					SOAPEnvelope env = getMsgQueue().take();
+
+					assertNotNull(env.getBody());
+					assertNotNull(env.getBody().getChildrenWithLocalName("msg"));
+
+					OMElement element = (OMElement) env.getBody()
+							.getChildrenWithLocalName("msg").next();
+
+					String text = element.getText();
+
+					if (text.indexOf(new Long(value).toString()) > 0) {
+						break;
+					}
+
+					fail("round trip of message failed"
+							+ " - due to invalid messege content");
+
+				} catch (InterruptedException e) {
+					fail("interrupted while waiting for message");
+				}
+			}
+
+			try {
+				clientApi.unSubscribe(brokerEPR, subscriptionID, null);
+			} catch (AxisFault e) {
+				// TODO Auto-generated catch block
+				e.printStackTrace();
+			}
+			clientApi.shutdownConsumerService();
+
+		} catch (AxisFault e) {
+			e.printStackTrace();
+			fail("unexpected exception occured");
+		}
+
+	}
+
+	@Test
+	public final void testMessageCount() {
+
+		try {
+			// messages might be returned out of sequence;
+
+			Integer totalMsgCount = 2;
+
+			List<Integer> recievedSequences = new LinkedList<Integer>();
+			final String MSG_TAG = "msg";
+			String msgFormat = "<msg>%d</msg>";
+			String topic = "WseRoundTripMsgCountTestTopic";
+
+			WseClientAPI clientApi = new WseClientAPI();
+
+			int consumerPort = new Integer(configs
+					.getProperty(ConfigKeys.CONSUMER_PORT));
+
+			String brokerEPR = configs
+					.getProperty(ConfigKeys.BROKER_EVENTING_SERVICE_EPR);
+			String[] consumerEPRs = clientApi.startConsumerService(
+					consumerPort, this);
+
+			assertTrue("consumer should have at least one epr",
+					consumerEPRs.length > 0);
+
+			String subscriptionID = clientApi.subscribe(brokerEPR,
+					consumerEPRs[0], topic);
+
+			for (int i = 1; i <= totalMsgCount; i++) {
+
+				String currentMsg = String.format(msgFormat, i);
+				clientApi.publish(brokerEPR, topic, currentMsg);
+			}
+
+			while (recievedSequences.size() < totalMsgCount) {
+				try {
+					SOAPEnvelope env = getMsgQueue().take();
+
+					assertNotNull(env.getBody());
+					assertNotNull(env.getBody().getChildrenWithLocalName(
+							MSG_TAG));
+
+					OMElement element = (OMElement) env.getBody()
+							.getChildrenWithLocalName(MSG_TAG).next();
+
+					int currentSequence = Integer.parseInt(element.getText());
+
+					assertFalse("duplicate message recieved", recievedSequences
+							.contains(currentSequence));
+
+					recievedSequences.add(currentSequence);
+
+				} catch (InterruptedException e) {
+					fail("interrupted while waiting for message");
+				}
+			}
+
+			try {
+				Thread.sleep(5000);
+
+				assertTrue("duplicate and/or invalid messages recieved",
+						getMsgQueue().isEmpty());
+
+			} catch (InterruptedException e) {
+				fail("test case was interrupted");
+			}
+
+			try {
+				clientApi.unSubscribe(brokerEPR, subscriptionID, null);
+			} catch (Exception e) {
+				// TODO Auto-generated catch block
+				e.printStackTrace();
+			}
+			clientApi.shutdownConsumerService();
+
+		} catch (AxisFault e) {
+			e.printStackTrace();
+			fail("unexpected exception occured");
+		}
+
+	}
+
+}

Added: incubator/airavata/donations/ogce-donation/modules/ws-messaging/broker/src/test/java/wsmg/client/WseClientAPISubscriptionsTest.java
URL: http://svn.apache.org/viewvc/incubator/airavata/donations/ogce-donation/modules/ws-messaging/broker/src/test/java/wsmg/client/WseClientAPISubscriptionsTest.java?rev=1142350&view=auto
==============================================================================
--- incubator/airavata/donations/ogce-donation/modules/ws-messaging/broker/src/test/java/wsmg/client/WseClientAPISubscriptionsTest.java (added)
+++ incubator/airavata/donations/ogce-donation/modules/ws-messaging/broker/src/test/java/wsmg/client/WseClientAPISubscriptionsTest.java Sun Jul  3 00:29:24 2011
@@ -0,0 +1,312 @@
+/*
+ * Licensed to the Lanka Software Foundation (LSF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The LSF 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 wsmg.client;
+
+import java.net.URL;
+import java.util.Properties;
+
+import junit.framework.TestCase;
+
+import org.apache.axis2.AxisFault;
+import org.apache.axis2.addressing.EndpointReference;
+import org.junit.After;
+import org.junit.AfterClass;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import wsmg.util.ConfigKeys;
+
+public class WseClientAPISubscriptionsTest extends TestCase {
+
+	static Properties configs = new Properties();
+
+	@BeforeClass
+	public static void setUpBeforeClass() throws Exception {
+
+	}
+
+	@AfterClass
+	public static void tearDownAfterClass() throws Exception {
+	}
+
+	@Before
+	public void setUp() throws Exception {
+		URL configURL = ClassLoader
+				.getSystemResource(ConfigKeys.CONFIG_FILE_NAME);
+		configs.load(configURL.openStream());
+
+	}
+
+	@After
+	public void tearDown() throws Exception {
+	}
+
+	// basic subscription test
+	@Test
+	public final void testSubscribe() {
+
+		try {
+			WseClientAPI clientApi = new WseClientAPI();
+
+			String brokerEPR = configs
+					.getProperty(ConfigKeys.BROKER_EVENTING_SERVICE_EPR);
+			String consumerEPR = configs.getProperty(ConfigKeys.CONSUMER_EPR);
+			String topic = "SimpleTopicWSE";
+
+			String subscriptionID = clientApi.subscribe(brokerEPR, consumerEPR,
+					topic);
+
+			assertNotNull(subscriptionID);
+
+		} catch (AxisFault e) {
+			e.printStackTrace();
+			fail("unable to subscribe: " + e.toString());
+		}
+
+	}
+
+	@Test
+	public final void testSubscribeWithURLBasedTopic() {
+
+		try {
+			WseClientAPI clientApi = new WseClientAPI();
+
+			String brokerEPR = configs
+					.getProperty(ConfigKeys.BROKER_EVENTING_SERVICE_EPR);
+			String consumerEPR = configs.getProperty(ConfigKeys.CONSUMER_EPR);
+			String topic = "MytopicFromURLWSE";
+
+			EndpointReference brokerEprWithTopic = WseClientAPI
+					.createEndpointReference(brokerEPR, topic);
+
+			String subscriptionID = clientApi.subscribe(brokerEprWithTopic
+					.getAddress(), consumerEPR, null);
+
+			assertNotNull(subscriptionID);
+
+		} catch (AxisFault e) {
+			e.printStackTrace();
+			fail("unable to subscribe: " + e.toString());
+		}
+
+	}
+
+	// check for duplicate subscription
+	@Test
+	public final void testSubscribeSameSubcription() {
+
+		try {
+			WseClientAPI clientApi = new WseClientAPI();
+
+			String brokerEPR = configs
+					.getProperty(ConfigKeys.BROKER_EVENTING_SERVICE_EPR);
+			String consumerEPR = configs.getProperty(ConfigKeys.CONSUMER_EPR);
+			String topic = "DuplicateSubscriptionTestWSE";
+
+			String subscriptionIDFirst = clientApi.subscribe(brokerEPR,
+					consumerEPR, topic);
+
+			assertNotNull(subscriptionIDFirst);
+
+			String subscriptionIDSecond = clientApi.subscribe(brokerEPR,
+					consumerEPR, topic);
+
+			assertEquals(subscriptionIDFirst, subscriptionIDSecond);
+
+		} catch (AxisFault e) {
+			e.printStackTrace();
+			fail("unable to subscribe: " + e.toString());
+		}
+
+	}
+
+	// check for Xpath subscription
+	@Test
+	public final void testSubscribeXpath() {
+
+		try {
+			WseClientAPI clientApi = new WseClientAPI();
+
+			String brokerEPR = configs
+					.getProperty(ConfigKeys.BROKER_EVENTING_SERVICE_EPR);
+			String consumerEPR = configs.getProperty(ConfigKeys.CONSUMER_EPR);
+			String topicExpression = null;
+			String xpathExpression = "/test/wse";
+
+			clientApi.subscribe(brokerEPR, consumerEPR, topicExpression,
+					xpathExpression);
+			String subscriptionID = clientApi.subscribe(brokerEPR, consumerEPR,
+					topicExpression);
+
+			assertNotNull(subscriptionID);
+
+		} catch (AxisFault e) {
+			e.printStackTrace();
+			fail("unable to subscribe: " + e.toString());
+		}
+
+	}
+
+	// check for Xpath subscription
+	@Test
+	public final void testSubscribeTopicAndXpath() {
+
+		try {
+			WseClientAPI clientApi = new WseClientAPI();
+
+			String brokerEPR = configs
+					.getProperty(ConfigKeys.BROKER_EVENTING_SERVICE_EPR);
+			String consumerEPR = configs.getProperty(ConfigKeys.CONSUMER_EPR);
+			String topicExpression = "testTopicAndXpathWSE";
+			String xpathExpression = "/test/wse1";
+
+			String subscriptionID = clientApi.subscribe(brokerEPR, consumerEPR,
+					topicExpression, xpathExpression);
+
+			assertNotNull(subscriptionID);
+
+		} catch (AxisFault e) {
+			e.printStackTrace();
+			fail("unable to subscribe: " + e.toString());
+		}
+
+	}
+
+	@Test
+	public final void testSubscribeURLTopicAndXpath() {
+
+		try {
+			WseClientAPI clientApi = new WseClientAPI();
+
+			String brokerEPR = configs
+					.getProperty(ConfigKeys.BROKER_EVENTING_SERVICE_EPR);
+			String consumerEPR = configs.getProperty(ConfigKeys.CONSUMER_EPR);
+			String topicExpression = "testTopicURLAndXpathWSE";
+			String xpathExpression = "/wse/xpath";
+
+			EndpointReference brokerEprWithTopic = WseClientAPI
+					.createEndpointReference(brokerEPR, topicExpression);
+
+			String subscriptionID = clientApi.subscribe(brokerEPR, consumerEPR,
+					null, xpathExpression);
+
+			assertNotNull(subscriptionID);
+
+		} catch (AxisFault e) {
+			e.printStackTrace();
+			fail("unable to subscribe: " + e.toString());
+		}
+
+	}
+
+	@Test
+	public final void testUnsubscribeSimpleTopic() {
+		try {
+			WseClientAPI clientApi = new WseClientAPI();
+
+			String brokerEPR = configs
+					.getProperty(ConfigKeys.BROKER_EVENTING_SERVICE_EPR);
+			String consumerEPR = configs.getProperty(ConfigKeys.CONSUMER_EPR);
+			String topicExpression = "testUnsubscribeTopicWSE";
+			String xpathExpression = null;
+
+			String subscriptionID = clientApi.subscribe(brokerEPR, consumerEPR,
+					topicExpression, xpathExpression);
+
+			assertNotNull(subscriptionID);
+
+			try {
+				clientApi.unSubscribe(brokerEPR, subscriptionID, null);
+			} catch (AxisFault e) {
+				// TODO Auto-generated catch block
+				e.printStackTrace();
+			}
+
+		} catch (AxisFault e) {
+			e.printStackTrace();
+			fail("unable to subscribe: " + e.toString());
+		}
+
+	}
+
+	@Test
+	public final void testUnsubscribeXpathTopic() {
+		try {
+			WseClientAPI clientApi = new WseClientAPI();
+
+			String brokerEPR = configs
+					.getProperty(ConfigKeys.BROKER_EVENTING_SERVICE_EPR);
+			String consumerEPR = configs.getProperty(ConfigKeys.CONSUMER_EPR);
+			String topicExpression = null;
+			String xpathExpression = "/www/xpath/wse";
+
+			String subscriptionID = clientApi.subscribe(brokerEPR, consumerEPR,
+					topicExpression, xpathExpression);
+
+			assertNotNull(subscriptionID);
+
+			try {
+				clientApi.unSubscribe(brokerEPR, subscriptionID, null);
+			} catch (Exception e) {
+				// TODO Auto-generated catch block
+				e.printStackTrace();
+			}
+
+		} catch (AxisFault e) {
+			e.printStackTrace();
+			fail("unable to subscribe: " + e.toString());
+		}
+
+	}
+
+	@Test
+	public final void testUnsubscribeSimpleAndXpathTopic() {
+		try {
+			WseClientAPI clientApi = new WseClientAPI();
+
+			String brokerEPR = configs
+					.getProperty(ConfigKeys.BROKER_EVENTING_SERVICE_EPR);
+			String consumerEPR = configs.getProperty(ConfigKeys.CONSUMER_EPR);
+			String topicExpression = "unsubscribeSimpleAndTopicTestWSE";
+			String xpathExpression = "/wse/test/xpath";
+
+			String subscriptionID = clientApi.subscribe(brokerEPR, consumerEPR,
+					topicExpression, xpathExpression);
+
+			assertNotNull(subscriptionID);
+
+			try {
+				clientApi.unSubscribe(brokerEPR, subscriptionID, null);
+			} catch (Exception e) {
+				// TODO Auto-generated catch block
+				e.printStackTrace();
+			}
+
+		} catch (AxisFault e) {
+			e.printStackTrace();
+			fail("unable to subscribe: " + e.toString());
+		}
+
+	}
+
+}

Added: incubator/airavata/donations/ogce-donation/modules/ws-messaging/broker/src/test/java/wsmg/client/WseClientAPITest.java
URL: http://svn.apache.org/viewvc/incubator/airavata/donations/ogce-donation/modules/ws-messaging/broker/src/test/java/wsmg/client/WseClientAPITest.java?rev=1142350&view=auto
==============================================================================
--- incubator/airavata/donations/ogce-donation/modules/ws-messaging/broker/src/test/java/wsmg/client/WseClientAPITest.java (added)
+++ incubator/airavata/donations/ogce-donation/modules/ws-messaging/broker/src/test/java/wsmg/client/WseClientAPITest.java Sun Jul  3 00:29:24 2011
@@ -0,0 +1,82 @@
+/*
+ * Licensed to the Lanka Software Foundation (LSF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The LSF 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 wsmg.client;
+
+import java.net.URL;
+import java.util.Properties;
+
+import junit.framework.TestCase;
+
+import org.apache.axis2.AxisFault;
+import org.junit.After;
+import org.junit.AfterClass;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Ignore;
+import org.junit.Test;
+
+@Ignore
+public class WseClientAPITest extends TestCase {
+
+	static class ConfigKeys {
+		static String BROKER_EPR = "broker.epr";
+		static String CONSUMER_EPR = "consumer.location";
+		static String CONSUMER_PORT = "consumer.port";
+		static String TOPIC_SIME = "topic.simple";
+		static String TOPIC_XPATH = "topic.xpath";
+	}
+
+	static Properties configs = new Properties();
+
+	@BeforeClass
+	public static void setUpBeforeClass() throws Exception {
+
+		URL configURL = ClassLoader.getSystemResource("unit_tests.properties");
+		configs.load(configURL.openStream());
+
+	}
+
+	@AfterClass
+	public static void tearDownAfterClass() throws Exception {
+	}
+
+	@Before
+	public void setUp() throws Exception {
+	}
+
+	@After
+	public void tearDown() throws Exception {
+	}
+
+	@Test
+	public final void testSubscribe() {
+		WseClientAPI clientApi = new WseClientAPI();
+		try {
+			clientApi.subscribe(configs.getProperty(ConfigKeys.BROKER_EPR), configs.getProperty(ConfigKeys.CONSUMER_EPR), configs.getProperty(ConfigKeys.TOPIC_SIME));
+		} catch (AxisFault e) {
+
+			fail("unable to subscribe: " + e.toString());
+		}
+
+	}
+
+}

Added: incubator/airavata/donations/ogce-donation/modules/ws-messaging/broker/src/test/java/wsmg/client/WsntClientAPIConsumerTest.java
URL: http://svn.apache.org/viewvc/incubator/airavata/donations/ogce-donation/modules/ws-messaging/broker/src/test/java/wsmg/client/WsntClientAPIConsumerTest.java?rev=1142350&view=auto
==============================================================================
--- incubator/airavata/donations/ogce-donation/modules/ws-messaging/broker/src/test/java/wsmg/client/WsntClientAPIConsumerTest.java (added)
+++ incubator/airavata/donations/ogce-donation/modules/ws-messaging/broker/src/test/java/wsmg/client/WsntClientAPIConsumerTest.java Sun Jul  3 00:29:24 2011
@@ -0,0 +1,100 @@
+/*
+ * Licensed to the Lanka Software Foundation (LSF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The LSF 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 wsmg.client;
+
+import java.net.URL;
+import java.util.Properties;
+
+import javax.xml.stream.XMLStreamException;
+
+import junit.framework.TestCase;
+
+import org.apache.axiom.soap.SOAPEnvelope;
+import org.apache.axis2.AxisFault;
+import org.junit.After;
+import org.junit.AfterClass;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import wsmg.util.ConfigKeys;
+
+
+
+public class WsntClientAPIConsumerTest extends TestCase {
+
+	static Properties configs = new Properties();
+
+	@BeforeClass
+	public static void setUpBeforeClass() throws Exception {
+
+	}
+
+	@AfterClass
+	public static void tearDownAfterClass() throws Exception {
+	}
+
+	@Before
+	public void setUp() throws Exception {
+		URL configURL = ClassLoader
+				.getSystemResource(ConfigKeys.CONFIG_FILE_NAME);
+		configs.load(configURL.openStream());
+
+	}
+
+	@After
+	public void tearDown() throws Exception {
+	}
+
+	// basic subscription test
+	@Test
+	public final void testConsumerService() {
+
+		ConsumerNotificationHandler handler = new ConsumerNotificationHandler() {
+
+			public void handleNotification(SOAPEnvelope msgEnvelope) {
+				try {
+					System.out.println(msgEnvelope.toStringWithConsume());
+				} catch (XMLStreamException e) {
+					// TODO Auto-generated catch block
+					e.printStackTrace();
+				}
+			}
+		};
+
+		try {
+			WsntClientAPI clientApi = new WsntClientAPI();
+
+			int consumerPort = new Integer(configs
+					.getProperty(ConfigKeys.CONSUMER_PORT));
+
+			clientApi.startConsumerService(consumerPort, handler);
+			clientApi.shutdownConsumerService();
+			
+		} catch (AxisFault e) {
+			e.printStackTrace();
+			fail("unable to start consumer service: " + e.toString());
+		}
+
+	}
+
+}

Added: incubator/airavata/donations/ogce-donation/modules/ws-messaging/broker/src/test/java/wsmg/client/WsntClientAPIPublishTest.java
URL: http://svn.apache.org/viewvc/incubator/airavata/donations/ogce-donation/modules/ws-messaging/broker/src/test/java/wsmg/client/WsntClientAPIPublishTest.java?rev=1142350&view=auto
==============================================================================
--- incubator/airavata/donations/ogce-donation/modules/ws-messaging/broker/src/test/java/wsmg/client/WsntClientAPIPublishTest.java (added)
+++ incubator/airavata/donations/ogce-donation/modules/ws-messaging/broker/src/test/java/wsmg/client/WsntClientAPIPublishTest.java Sun Jul  3 00:29:24 2011
@@ -0,0 +1,106 @@
+/*
+ * Licensed to the Lanka Software Foundation (LSF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The LSF 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 wsmg.client;
+
+import java.net.URL;
+import java.util.Properties;
+
+import junit.framework.TestCase;
+
+import org.apache.axis2.AxisFault;
+import org.junit.After;
+import org.junit.AfterClass;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import wsmg.util.ConfigKeys;
+
+public class WsntClientAPIPublishTest extends TestCase {
+
+	static Properties configs = new Properties();
+
+	@BeforeClass
+	public static void setUpBeforeClass() throws Exception {
+
+	}
+
+	@AfterClass
+	public static void tearDownAfterClass() throws Exception {
+	}
+
+	@Before
+	public void setUp() throws Exception {
+		URL configURL = ClassLoader
+				.getSystemResource(ConfigKeys.CONFIG_FILE_NAME);
+		configs.load(configURL.openStream());
+
+	}
+
+	@After
+	public void tearDown() throws Exception {
+	}
+
+	// basic pubilsh test
+	@Test
+	public final void testPublishWithBrokerURL() {
+
+		try {
+			WsntClientAPI clientApi = new WsntClientAPI();
+			
+			String brokerEPR = configs
+					.getProperty(ConfigKeys.BROKER_NOTIFICATIONS_SERVICE_EPR);
+			String topic = "MYSimpleTOPIC";
+
+			String message = "<msg> test message </msg>";
+			// should throw an exception
+			clientApi.publish(brokerEPR, topic, message);
+			
+
+		} catch (AxisFault e) {
+			e.printStackTrace();
+			fail("unable to publish: " + e.toString());
+		}
+
+	}
+
+	// basic publish test
+	@Test(expected = AxisFault.class)
+	public final void testPublishInvalidBrokerLocation() {
+
+		WseClientAPI clientApi = new WseClientAPI();
+
+		String brokerEPR = "http://invalid/url/";
+		String topic = "MYSimpleTOPIC";
+
+		String message = "<msg> test message </msg>";
+		// should throw an exception
+		try {
+			clientApi.publish(brokerEPR, topic, message);
+			fail("didn't throw an exception");
+		} catch (AxisFault e) {
+			// TODO Auto-generated catch block
+			e.printStackTrace();
+		}
+
+	}
+}

Added: incubator/airavata/donations/ogce-donation/modules/ws-messaging/broker/src/test/java/wsmg/client/WsntClientAPIRoundTripTests.java
URL: http://svn.apache.org/viewvc/incubator/airavata/donations/ogce-donation/modules/ws-messaging/broker/src/test/java/wsmg/client/WsntClientAPIRoundTripTests.java?rev=1142350&view=auto
==============================================================================
--- incubator/airavata/donations/ogce-donation/modules/ws-messaging/broker/src/test/java/wsmg/client/WsntClientAPIRoundTripTests.java (added)
+++ incubator/airavata/donations/ogce-donation/modules/ws-messaging/broker/src/test/java/wsmg/client/WsntClientAPIRoundTripTests.java Sun Jul  3 00:29:24 2011
@@ -0,0 +1,260 @@
+/*
+ * Licensed to the Lanka Software Foundation (LSF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The LSF 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 wsmg.client;
+
+import java.net.URL;
+import java.util.Iterator;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Properties;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+
+import javax.xml.namespace.QName;
+
+import junit.framework.TestCase;
+
+import org.apache.axiom.om.OMElement;
+import org.apache.axiom.soap.SOAPEnvelope;
+import org.apache.axis2.AxisFault;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+import wsmg.commons.WsmgNameSpaceConstants;
+import wsmg.util.ConfigKeys;
+
+/**
+ * @author ramith
+ * 
+ */
+public class WsntClientAPIRoundTripTests extends TestCase implements
+		ConsumerNotificationHandler {
+
+	static Properties configs = new Properties();
+
+	BlockingQueue<SOAPEnvelope> queue = new LinkedBlockingQueue<SOAPEnvelope>();
+
+	public void handleNotification(SOAPEnvelope msgEnvelope) {
+
+		queue.add(msgEnvelope);
+		System.out.println(msgEnvelope);
+	}
+
+	BlockingQueue<SOAPEnvelope> getMsgQueue() {
+		return queue;
+	}
+
+	/**
+	 * @throws java.lang.Exception
+	 */
+	@Before
+	public void setUp() throws Exception {
+		URL configURL = ClassLoader
+				.getSystemResource(ConfigKeys.CONFIG_FILE_NAME);
+		configs.load(configURL.openStream());
+	}
+
+	/**
+	 * @throws java.lang.Exception
+	 */
+	@After
+	public void tearDown() throws Exception {
+	}
+
+	@Test
+	public final void testRoundTrip() {
+
+		try {
+
+			long value = System.currentTimeMillis();
+			String msg = String.format("<msg> current time is : %d </msg>",
+					value);
+
+			WsntClientAPI clientApi = new WsntClientAPI();
+
+			int consumerPort = new Integer(configs
+					.getProperty(ConfigKeys.CONSUMER_PORT));
+
+			String brokerEPR = configs
+					.getProperty(ConfigKeys.BROKER_NOTIFICATIONS_SERVICE_EPR);
+			String[] consumerEPRs = clientApi.startConsumerService(
+					consumerPort, this);
+
+			assertTrue(consumerEPRs.length > 0);
+
+			String topic = "WsntRoundTripTestTopic";
+
+			String subscriptionID = clientApi.subscribe(brokerEPR,
+					consumerEPRs[0], topic);
+
+			clientApi.publish(brokerEPR, topic, msg);
+
+			while (true) {
+				try {
+					SOAPEnvelope env = getMsgQueue().take();
+
+					OMElement element = extractRawMessage(env);
+					String text = element.getText();
+
+					if (text.indexOf(new Long(value).toString()) > 0) {
+						break;
+					}
+
+					fail("round trip of message failed"
+							+ " - due to invalid messege content");
+
+				} catch (InterruptedException e) {
+					fail("interrupted while waiting for message");
+				}
+			}
+
+			try {
+				clientApi.unSubscribe(brokerEPR, subscriptionID, null);
+			} catch (AxisFault e) {
+				// TODO Auto-generated catch block
+				e.printStackTrace();
+			}
+			clientApi.shutdownConsumerService();
+			
+		} catch (AxisFault e) {
+			e.printStackTrace();
+			fail("unexpected exception occured");
+		}
+
+	}
+
+	@Test
+	public final void testMessageCount() {
+
+		try {
+			// messages might be returned out of sequence;
+
+			Integer totalMsgCount = 2;
+
+			List<Integer> recievedSequences = new LinkedList<Integer>();
+			final String MSG_TAG = "msg";
+			String msgFormat = "<msg>%d</msg>";
+			String topic = "WsntRoundTripMsgCountTestTopic";
+
+			WsntClientAPI clientApi = new WsntClientAPI();
+
+			int consumerPort = new Integer(configs
+					.getProperty(ConfigKeys.CONSUMER_PORT));
+
+			String brokerEPR = configs
+					.getProperty(ConfigKeys.BROKER_NOTIFICATIONS_SERVICE_EPR);
+			String[] consumerEPRs = clientApi.startConsumerService(
+					consumerPort, this);
+
+			assertTrue("consumer should have at least one epr",
+					consumerEPRs.length > 0);
+
+			String subscriptionID = clientApi.subscribe(brokerEPR,
+					consumerEPRs[0], topic);
+
+			for (int i = 1; i <= totalMsgCount; i++) {
+
+				String currentMsg = String.format(msgFormat, i);
+				clientApi.publish(brokerEPR, topic, currentMsg);
+			}
+
+			while (recievedSequences.size() < totalMsgCount) {
+				try {
+					SOAPEnvelope env = getMsgQueue().take();
+
+					OMElement element = extractRawMessage(env);
+
+					int currentSequence = Integer.parseInt(element.getText());
+
+					assertFalse("duplicate message recieved", recievedSequences
+							.contains(currentSequence));
+
+					recievedSequences.add(currentSequence);
+
+				} catch (InterruptedException e) {
+					fail("interrupted while waiting for message");
+				}
+			}
+
+			try {
+				Thread.sleep(5000);
+
+				assertTrue("duplicate and/or invalid messages recieved",
+						getMsgQueue().isEmpty());
+
+			} catch (InterruptedException e) {
+				fail("test case was interrupted");
+			}
+
+			try {
+				clientApi.unSubscribe(brokerEPR, subscriptionID, null);
+			} catch (AxisFault e) {
+				// TODO Auto-generated catch block
+				e.printStackTrace();
+			}
+			clientApi.shutdownConsumerService();
+			
+		} catch (AxisFault e) {
+			e.printStackTrace();
+			fail("unexpected exception occured");
+		}
+
+	}
+
+	private OMElement extractRawMessage(SOAPEnvelope env) {
+
+		QName notify = new QName(WsmgNameSpaceConstants.WSNT_NS
+				.getNamespaceURI(), "Notify", WsmgNameSpaceConstants.WSNT_NS
+				.getPrefix());
+
+		QName notifyMsg = new QName(WsmgNameSpaceConstants.WSNT_NS
+				.getNamespaceURI(), "NotificationMessage",
+				WsmgNameSpaceConstants.WSNT_NS.getPrefix());
+
+		QName msg = new QName(WsmgNameSpaceConstants.WSNT_NS.getNamespaceURI(),
+				"Message", WsmgNameSpaceConstants.WSNT_NS.getPrefix());
+
+		assertNotNull("envelope can't be null", env);
+		assertNotNull("envelope don't have a body", env.getBody());
+
+		Iterator ite = env.getBody().getChildrenWithName(notify);
+
+		assertTrue(notify.getLocalPart() + " tag is not found", ite.hasNext());
+
+		OMElement ele = (OMElement) ite.next();
+		ite = ele.getChildrenWithName(notifyMsg);
+
+		assertTrue(notifyMsg.getLocalPart() + " tag is not found", ite
+				.hasNext());
+
+		ele = (OMElement) ite.next();
+
+		ite = ele.getChildrenWithName(msg);
+		assertTrue(msg.getLocalPart() + " tag is not found", ite.hasNext());
+
+		ele = (OMElement) ite.next();
+
+		assertNotNull("raw message was not found", ele.getFirstElement());
+
+		return ele.getFirstElement();
+	}
+
+}