trunk/activemq-core/src/test/java/org/apache/activemq/network/NetworkLoadTest.java - activemq - Git at Google

 /**
  * Licensed to the Apache Software Foundation (ASF) under one or more
  * contributor license agreements.  See the NOTICE file distributed with
  * this work for additional information regarding copyright ownership.
  * The ASF licenses this file to You under the Apache License, Version 2.0
  * (the "License"); you may not use this file except in compliance with
  * the License.  You may obtain a copy of the License at
  *
  *      http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
 package org.apache.activemq.network;

 import java.net.URI;
 import java.util.ArrayList;
 import java.util.Date;
 import java.util.List;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.atomic.AtomicReference;

 import javax.jms.Connection;
 import javax.jms.DeliveryMode;
 import javax.jms.JMSException;
 import javax.jms.Message;
 import javax.jms.MessageConsumer;
 import javax.jms.MessageListener;
 import javax.jms.MessageProducer;
 import javax.jms.Session;
 import javax.jms.TextMessage;

 import junit.framework.TestCase;

 import org.apache.activemq.ActiveMQConnectionFactory;
 import org.apache.activemq.ActiveMQPrefetchPolicy;
 import org.apache.activemq.broker.BrokerService;
 import org.apache.activemq.broker.TransportConnector;
 import org.apache.activemq.broker.region.policy.NoSubscriptionRecoveryPolicy;
 import org.apache.activemq.broker.region.policy.PolicyEntry;
 import org.apache.activemq.broker.region.policy.PolicyMap;
 import org.apache.activemq.broker.region.policy.VMPendingQueueMessageStoragePolicy;
 import org.apache.activemq.broker.region.policy.VMPendingSubscriberMessageStoragePolicy;
 import org.apache.activemq.command.ActiveMQQueue;
 import org.apache.activemq.command.ActiveMQTextMessage;
 import org.apache.activemq.usage.SystemUsage;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;

 /**
  * This test case is used to load test store and forwarding between brokers.  It sets up
  * n brokers to which have a chain of queues which this test consumes and produces to.
  *
  * If the network bridges gets stuck at any point subsequent queues will not get messages.  This test
  * samples the production and consumption stats every second and if the flow of messages
  * get stuck then this tast fails.  The test monitors the flow of messages for 1 min.
  *
  * @author chirino
  */
 public class NetworkLoadTest extends TestCase {

 	private static final transient Logger LOG = LoggerFactory.getLogger(NetworkLoadTest.class);

 	// How many times do we sample?
     private static final long SAMPLES = Integer.parseInt(System.getProperty("SAMPLES", ""+60*1/5));
     // Slower machines might need to make this bigger.
     private static final long SAMPLE_DURATION = Integer.parseInt(System.getProperty("SAMPLES_DURATION", "" + 1000 * 5));
 	protected static final int BROKER_COUNT = 4;
 	protected static final int MESSAGE_SIZE = 2000;
         String groupId;

 	class ForwardingClient {

 		private final AtomicLong forwardCounter = new AtomicLong();
 		private final Connection toConnection;
 		private final Connection fromConnection;

 		public ForwardingClient(int from, int to) throws JMSException {
 			toConnection = createConnection(from);
 			Session toSession = toConnection.createSession(false, Session.AUTO_ACKNOWLEDGE);
 			final MessageProducer producer = toSession.createProducer(new ActiveMQQueue("Q"+to));
 			producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
 			producer.setDisableMessageID(true);

 			fromConnection = createConnection(from);
 			Session fromSession = fromConnection.createSession(false, Session.AUTO_ACKNOWLEDGE);
 			MessageConsumer consumer = fromSession.createConsumer(new ActiveMQQueue("Q"+from));

 			consumer.setMessageListener(new MessageListener() {
 					public void onMessage(Message msg) {
 						try {
 							producer.send(msg);
 							forwardCounter.incrementAndGet();
 						} catch (JMSException e) {
 							// this is caused by the connection getting closed.
 						}
 					}
 			});
 		}

 		public void start() throws JMSException {
 			toConnection.start();
 			fromConnection.start();
 		}

 		public void stop() throws JMSException {
 		        toConnection.stop();
 			fromConnection.stop();
 		}

 		public void close() throws JMSException {
 			toConnection.close();
 			fromConnection.close();
 		}
 	}

 	private BrokerService[] brokers;
 	private ForwardingClient[] forwardingClients;


 	protected void setUp() throws Exception {
 	        groupId = "network-load-test-"+System.currentTimeMillis();
 		brokers = new BrokerService[BROKER_COUNT];
 		for (int i = 0; i < brokers.length; i++) {
 		    LOG.info("Starting broker: "+i);
 			brokers[i] = createBroker(i);
 			brokers[i].start();
 		}

 		// Wait for the network connection to get setup.
 		// The wait is exponential since every broker has to connect to every other broker.
 		Thread.sleep(BROKER_COUNT*BROKER_COUNT*50);

 		forwardingClients = new ForwardingClient[BROKER_COUNT-1];
 		for (int i = 0; i < forwardingClients.length; i++) {
 		    LOG.info("Starting fowarding client "+i);
 			forwardingClients[i] = new ForwardingClient(i, i+1);
 			forwardingClients[i].start();
 		}
 	}

 	protected void tearDown() throws Exception {
 		for (int i = 0; i < forwardingClients.length; i++) {
 		    LOG.info("Stoping fowarding client "+i);
 			forwardingClients[i].close();
 		}
 		for (int i = 0; i < brokers.length; i++) {
 		    LOG.info("Stoping broker "+i);
 			brokers[i].stop();
 		}
 	}

 	protected Connection createConnection(int brokerId) throws JMSException {
 		ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://localhost:"+(60000+brokerId));
         connectionFactory.setOptimizedMessageDispatch(true);
         connectionFactory.setCopyMessageOnSend(false);
         connectionFactory.setUseCompression(false);
         connectionFactory.setDispatchAsync(true);
         connectionFactory.setUseAsyncSend(false);
         connectionFactory.setOptimizeAcknowledge(false);
         connectionFactory.setWatchTopicAdvisories(false);
         ActiveMQPrefetchPolicy qPrefetchPolicy= new ActiveMQPrefetchPolicy();
         qPrefetchPolicy.setQueuePrefetch(100);
         qPrefetchPolicy.setTopicPrefetch(1000);
         connectionFactory.setPrefetchPolicy(qPrefetchPolicy);
         connectionFactory.setAlwaysSyncSend(true);
 		return connectionFactory.createConnection();
 	}

 	protected BrokerService createBroker(int brokerId) throws Exception {
 		BrokerService broker = new BrokerService();
 		broker.setBrokerName("broker-" + brokerId);
 		broker.setPersistent(false);
 		broker.setUseJmx(true);
 		broker.getManagementContext().setCreateConnector(false);

 		final SystemUsage memoryManager = new SystemUsage();
 		memoryManager.getMemoryUsage().setLimit(1024 * 1024 * 50); // 50 MB
 		broker.setSystemUsage(memoryManager);

 		final List<PolicyEntry> policyEntries = new ArrayList<PolicyEntry>();
 		final PolicyEntry entry = new PolicyEntry();
 		entry.setQueue(">");
 		entry.setMemoryLimit(1024 * 1024 * 1); // Set to 1 MB
 		entry.setPendingSubscriberPolicy(new VMPendingSubscriberMessageStoragePolicy());
 		entry.setPendingQueuePolicy(new VMPendingQueueMessageStoragePolicy());
 		policyEntries.add(entry);

 		// This is to turn of the default behavior of storing topic messages for retroactive consumption
 		final PolicyEntry topicPolicyEntry = new PolicyEntry();
 		topicPolicyEntry.setTopic(">");
 		final NoSubscriptionRecoveryPolicy noSubscriptionRecoveryPolicy = new NoSubscriptionRecoveryPolicy();
 		topicPolicyEntry.setSubscriptionRecoveryPolicy(noSubscriptionRecoveryPolicy);

 		final PolicyMap policyMap = new PolicyMap();
 		policyMap.setPolicyEntries(policyEntries);
 		broker.setDestinationPolicy(policyMap);

         TransportConnector transportConnector = new TransportConnector();
         transportConnector.setUri(new URI("tcp://localhost:"+(60000+brokerId)));

         transportConnector.setDiscoveryUri(new URI("multicast://default?group="+groupId));
         broker.addConnector(transportConnector);

         DiscoveryNetworkConnector networkConnector = new DiscoveryNetworkConnector();
         networkConnector.setUri(new URI("multicast://default?group="+groupId));
 	    networkConnector.setBridgeTempDestinations(true);
 	    networkConnector.setPrefetchSize(1);
 	    broker.addNetworkConnector(networkConnector);

 		return broker;
 	}

 	public void testRequestReply() throws Exception {

 		final int to = 0; // Send to the first broker
 		int from = brokers.length-1; // consume from the last broker..

 	    LOG.info("Staring Final Consumer");

 	    Connection fromConnection = createConnection(from);
 		fromConnection.start();
 		Session fromSession = fromConnection.createSession(false, Session.AUTO_ACKNOWLEDGE);
 		MessageConsumer consumer = fromSession.createConsumer(new ActiveMQQueue("Q"+from));

 		final AtomicReference<ActiveMQTextMessage> lastMessageReceived = new AtomicReference<ActiveMQTextMessage>();
 		final AtomicLong producedMessages = new AtomicLong();
 		final AtomicLong receivedMessages = new AtomicLong();
 		final AtomicBoolean done = new AtomicBoolean();

 		// Setup the consumer..
 		consumer.setMessageListener(new MessageListener() {
 			public void onMessage(Message msg) {
 				ActiveMQTextMessage m = (ActiveMQTextMessage) msg;
 				ActiveMQTextMessage last = lastMessageReceived.get();
 				if( last!=null ) {
 					// Some order checking...
 					if( last.getMessageId().getProducerSequenceId() > m.getMessageId().getProducerSequenceId() ) {
 						System.out.println("Received an out of order message. Got "+m.getMessageId()+", expected something after "+last.getMessageId());
 					}
 				}
 				lastMessageReceived.set(m);
 				receivedMessages.incrementAndGet();
 			}
 		});

 	    LOG.info("Staring Initial Producer");
 		final Connection toConnection = createConnection(to);
 		Thread producer = new Thread("Producer") {
 			@Override
 			public void run() {
 				try {
 					toConnection.start();
 					Session toSession = toConnection.createSession(false, Session.AUTO_ACKNOWLEDGE);
 					final MessageProducer producer = toSession.createProducer(new ActiveMQQueue("Q"+to));
 					producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
 					producer.setDisableMessageID(true);

 					for (int i = 0; !done.get(); i++) {
 						TextMessage msg = toSession.createTextMessage(createMessageText(i));
 						producer.send(msg);
 						producedMessages.incrementAndGet();
 					}
 				} catch (JMSException e) {
 					e.printStackTrace();
 				}
 			}

 		    private String createMessageText(int index) {
 				StringBuffer buffer = new StringBuffer(MESSAGE_SIZE);
 				buffer.append(index + " on " + new Date() + " ...");
 				if (buffer.length() > MESSAGE_SIZE) {
 					return buffer.substring(0, MESSAGE_SIZE);
 				}
 				for (int i = buffer.length(); i < MESSAGE_SIZE; i++) {
 					buffer.append(' ');
 				}

 				return buffer.toString();
 			}
 		};
 		producer.start();


 		// Give the forwarding clients a chance to get going and fill the down
 		// stream broker queues..
 		Thread.sleep(BROKER_COUNT*200);

         for (int i = 0; i < SAMPLES; i++) {

             long start = System.currentTimeMillis();
             producedMessages.set(0);
             receivedMessages.set(0);
             for (int j = 0; j < forwardingClients.length; j++) {
     			forwardingClients[j].forwardCounter.set(0);
     		}

             Thread.sleep(SAMPLE_DURATION);

             long end = System.currentTimeMillis();
             long r = receivedMessages.get();
             long p = producedMessages.get();

             LOG.info("published: " + p + " msgs at " + (p * 1000f / (end - start)) + " msgs/sec, " + "consumed: " + r + " msgs at " + (r * 1000f / (end - start)) + " msgs/sec");

             StringBuffer fwdingmsg = new StringBuffer(500);
             fwdingmsg.append("  forwarding counters: ");
             for (int j = 0; j < forwardingClients.length; j++) {
             	if( j!= 0 ) {
             		fwdingmsg.append(", ");
             	}
                 fwdingmsg.append(forwardingClients[j].forwardCounter.get());
     		}
             LOG.info(fwdingmsg.toString());

             // The test is just checking to make sure thaat the producer and consumer does not hang
             // due to the network hops take to route the message form the producer to the consumer.
             assertTrue("Recieved some messages since last sample", r>0);
             assertTrue("Produced some messages since last sample", p>0);

         }
         LOG.info("Sample done.");
         done.set(true);
         // Wait for the producer to finish.
         producer.join(1000*5);
         toConnection.close();
         fromConnection.close();

 	}


 }
	/**
	* Licensed to the Apache Software Foundation (ASF) under one or more
	* contributor license agreements. See the NOTICE file distributed with
	* this work for additional information regarding copyright ownership.
	* The ASF licenses this file to You under the Apache License, Version 2.0
	* (the "License"); you may not use this file except in compliance with
	* the License. You may obtain a copy of the License at
	*
	* http://www.apache.org/licenses/LICENSE-2.0
	*
	* Unless required by applicable law or agreed to in writing, software
	* distributed under the License is distributed on an "AS IS" BASIS,
	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
	* See the License for the specific language governing permissions and
	* limitations under the License.
	*/
	package org.apache.activemq.network;

	import java.net.URI;
	import java.util.ArrayList;
	import java.util.Date;
	import java.util.List;
	import java.util.concurrent.atomic.AtomicBoolean;
	import java.util.concurrent.atomic.AtomicLong;
	import java.util.concurrent.atomic.AtomicReference;

	import javax.jms.Connection;
	import javax.jms.DeliveryMode;
	import javax.jms.JMSException;
	import javax.jms.Message;
	import javax.jms.MessageConsumer;
	import javax.jms.MessageListener;
	import javax.jms.MessageProducer;
	import javax.jms.Session;
	import javax.jms.TextMessage;

	import junit.framework.TestCase;

	import org.apache.activemq.ActiveMQConnectionFactory;
	import org.apache.activemq.ActiveMQPrefetchPolicy;
	import org.apache.activemq.broker.BrokerService;
	import org.apache.activemq.broker.TransportConnector;
	import org.apache.activemq.broker.region.policy.NoSubscriptionRecoveryPolicy;
	import org.apache.activemq.broker.region.policy.PolicyEntry;
	import org.apache.activemq.broker.region.policy.PolicyMap;
	import org.apache.activemq.broker.region.policy.VMPendingQueueMessageStoragePolicy;
	import org.apache.activemq.broker.region.policy.VMPendingSubscriberMessageStoragePolicy;
	import org.apache.activemq.command.ActiveMQQueue;
	import org.apache.activemq.command.ActiveMQTextMessage;
	import org.apache.activemq.usage.SystemUsage;
	import org.slf4j.Logger;
	import org.slf4j.LoggerFactory;

	/**
	* This test case is used to load test store and forwarding between brokers. It sets up
	* n brokers to which have a chain of queues which this test consumes and produces to.
	*
	* If the network bridges gets stuck at any point subsequent queues will not get messages. This test
	* samples the production and consumption stats every second and if the flow of messages
	* get stuck then this tast fails. The test monitors the flow of messages for 1 min.
	*
	* @author chirino
	*/
	public class NetworkLoadTest extends TestCase {

	private static final transient Logger LOG = LoggerFactory.getLogger(NetworkLoadTest.class);

	// How many times do we sample?
	private static final long SAMPLES = Integer.parseInt(System.getProperty("SAMPLES", ""+60*1/5));
	// Slower machines might need to make this bigger.
	private static final long SAMPLE_DURATION = Integer.parseInt(System.getProperty("SAMPLES_DURATION", "" + 1000 * 5));
	protected static final int BROKER_COUNT = 4;
	protected static final int MESSAGE_SIZE = 2000;
	String groupId;

	class ForwardingClient {

	private final AtomicLong forwardCounter = new AtomicLong();
	private final Connection toConnection;
	private final Connection fromConnection;

	public ForwardingClient(int from, int to) throws JMSException {
	toConnection = createConnection(from);
	Session toSession = toConnection.createSession(false, Session.AUTO_ACKNOWLEDGE);
	final MessageProducer producer = toSession.createProducer(new ActiveMQQueue("Q"+to));
	producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
	producer.setDisableMessageID(true);

	fromConnection = createConnection(from);
	Session fromSession = fromConnection.createSession(false, Session.AUTO_ACKNOWLEDGE);
	MessageConsumer consumer = fromSession.createConsumer(new ActiveMQQueue("Q"+from));

	consumer.setMessageListener(new MessageListener() {
	public void onMessage(Message msg) {
	try {
	producer.send(msg);
	forwardCounter.incrementAndGet();
	} catch (JMSException e) {
	// this is caused by the connection getting closed.
	}
	}
	});
	}

	public void start() throws JMSException {
	toConnection.start();
	fromConnection.start();
	}

	public void stop() throws JMSException {
	toConnection.stop();
	fromConnection.stop();
	}

	public void close() throws JMSException {
	toConnection.close();
	fromConnection.close();
	}
	}

	private BrokerService[] brokers;
	private ForwardingClient[] forwardingClients;


	protected void setUp() throws Exception {
	groupId = "network-load-test-"+System.currentTimeMillis();
	brokers = new BrokerService[BROKER_COUNT];
	for (int i = 0; i < brokers.length; i++) {
	LOG.info("Starting broker: "+i);
	brokers[i] = createBroker(i);
	brokers[i].start();
	}

	// Wait for the network connection to get setup.
	// The wait is exponential since every broker has to connect to every other broker.
	Thread.sleep(BROKER_COUNTBROKER_COUNT50);

	forwardingClients = new ForwardingClient[BROKER_COUNT-1];
	for (int i = 0; i < forwardingClients.length; i++) {
	LOG.info("Starting fowarding client "+i);
	forwardingClients[i] = new ForwardingClient(i, i+1);
	forwardingClients[i].start();
	}
	}

	protected void tearDown() throws Exception {
	for (int i = 0; i < forwardingClients.length; i++) {
	LOG.info("Stoping fowarding client "+i);
	forwardingClients[i].close();
	}
	for (int i = 0; i < brokers.length; i++) {
	LOG.info("Stoping broker "+i);
	brokers[i].stop();
	}
	}

	protected Connection createConnection(int brokerId) throws JMSException {
	ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://localhost:"+(60000+brokerId));
	connectionFactory.setOptimizedMessageDispatch(true);
	connectionFactory.setCopyMessageOnSend(false);
	connectionFactory.setUseCompression(false);
	connectionFactory.setDispatchAsync(true);
	connectionFactory.setUseAsyncSend(false);
	connectionFactory.setOptimizeAcknowledge(false);
	connectionFactory.setWatchTopicAdvisories(false);
	ActiveMQPrefetchPolicy qPrefetchPolicy= new ActiveMQPrefetchPolicy();
	qPrefetchPolicy.setQueuePrefetch(100);
	qPrefetchPolicy.setTopicPrefetch(1000);
	connectionFactory.setPrefetchPolicy(qPrefetchPolicy);
	connectionFactory.setAlwaysSyncSend(true);
	return connectionFactory.createConnection();
	}

	protected BrokerService createBroker(int brokerId) throws Exception {
	BrokerService broker = new BrokerService();
	broker.setBrokerName("broker-" + brokerId);
	broker.setPersistent(false);
	broker.setUseJmx(true);
	broker.getManagementContext().setCreateConnector(false);

	final SystemUsage memoryManager = new SystemUsage();
	memoryManager.getMemoryUsage().setLimit(1024 * 1024 * 50); // 50 MB
	broker.setSystemUsage(memoryManager);

	final List<PolicyEntry> policyEntries = new ArrayList<PolicyEntry>();
	final PolicyEntry entry = new PolicyEntry();
	entry.setQueue(">");
	entry.setMemoryLimit(1024 * 1024 * 1); // Set to 1 MB
	entry.setPendingSubscriberPolicy(new VMPendingSubscriberMessageStoragePolicy());
	entry.setPendingQueuePolicy(new VMPendingQueueMessageStoragePolicy());
	policyEntries.add(entry);

	// This is to turn of the default behavior of storing topic messages for retroactive consumption
	final PolicyEntry topicPolicyEntry = new PolicyEntry();
	topicPolicyEntry.setTopic(">");
	final NoSubscriptionRecoveryPolicy noSubscriptionRecoveryPolicy = new NoSubscriptionRecoveryPolicy();
	topicPolicyEntry.setSubscriptionRecoveryPolicy(noSubscriptionRecoveryPolicy);

	final PolicyMap policyMap = new PolicyMap();
	policyMap.setPolicyEntries(policyEntries);
	broker.setDestinationPolicy(policyMap);

	TransportConnector transportConnector = new TransportConnector();
	transportConnector.setUri(new URI("tcp://localhost:"+(60000+brokerId)));

	transportConnector.setDiscoveryUri(new URI("multicast://default?group="+groupId));
	broker.addConnector(transportConnector);

	DiscoveryNetworkConnector networkConnector = new DiscoveryNetworkConnector();
	networkConnector.setUri(new URI("multicast://default?group="+groupId));
	networkConnector.setBridgeTempDestinations(true);
	networkConnector.setPrefetchSize(1);
	broker.addNetworkConnector(networkConnector);

	return broker;
	}

	public void testRequestReply() throws Exception {

	final int to = 0; // Send to the first broker
	int from = brokers.length-1; // consume from the last broker..

	LOG.info("Staring Final Consumer");

	Connection fromConnection = createConnection(from);
	fromConnection.start();
	Session fromSession = fromConnection.createSession(false, Session.AUTO_ACKNOWLEDGE);
	MessageConsumer consumer = fromSession.createConsumer(new ActiveMQQueue("Q"+from));

	final AtomicReference<ActiveMQTextMessage> lastMessageReceived = new AtomicReference<ActiveMQTextMessage>();
	final AtomicLong producedMessages = new AtomicLong();
	final AtomicLong receivedMessages = new AtomicLong();
	final AtomicBoolean done = new AtomicBoolean();

	// Setup the consumer..
	consumer.setMessageListener(new MessageListener() {
	public void onMessage(Message msg) {
	ActiveMQTextMessage m = (ActiveMQTextMessage) msg;
	ActiveMQTextMessage last = lastMessageReceived.get();
	if( last!=null ) {
	// Some order checking...
	if( last.getMessageId().getProducerSequenceId() > m.getMessageId().getProducerSequenceId() ) {
	System.out.println("Received an out of order message. Got "+m.getMessageId()+", expected something after "+last.getMessageId());
	}
	}
	lastMessageReceived.set(m);
	receivedMessages.incrementAndGet();
	}
	});

	LOG.info("Staring Initial Producer");
	final Connection toConnection = createConnection(to);
	Thread producer = new Thread("Producer") {
	@Override
	public void run() {
	try {
	toConnection.start();
	Session toSession = toConnection.createSession(false, Session.AUTO_ACKNOWLEDGE);
	final MessageProducer producer = toSession.createProducer(new ActiveMQQueue("Q"+to));
	producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
	producer.setDisableMessageID(true);

	for (int i = 0; !done.get(); i++) {
	TextMessage msg = toSession.createTextMessage(createMessageText(i));
	producer.send(msg);
	producedMessages.incrementAndGet();
	}
	} catch (JMSException e) {
	e.printStackTrace();
	}
	}

	private String createMessageText(int index) {
	StringBuffer buffer = new StringBuffer(MESSAGE_SIZE);
	buffer.append(index + " on " + new Date() + " ...");
	if (buffer.length() > MESSAGE_SIZE) {
	return buffer.substring(0, MESSAGE_SIZE);
	}
	for (int i = buffer.length(); i < MESSAGE_SIZE; i++) {
	buffer.append(' ');
	}

	return buffer.toString();
	}
	};
	producer.start();


	// Give the forwarding clients a chance to get going and fill the down
	// stream broker queues..
	Thread.sleep(BROKER_COUNT*200);

	for (int i = 0; i < SAMPLES; i++) {

	long start = System.currentTimeMillis();
	producedMessages.set(0);
	receivedMessages.set(0);
	for (int j = 0; j < forwardingClients.length; j++) {
	forwardingClients[j].forwardCounter.set(0);
	}

	Thread.sleep(SAMPLE_DURATION);

	long end = System.currentTimeMillis();
	long r = receivedMessages.get();
	long p = producedMessages.get();

	LOG.info("published: " + p + " msgs at " + (p * 1000f / (end - start)) + " msgs/sec, " + "consumed: " + r + " msgs at " + (r * 1000f / (end - start)) + " msgs/sec");

	StringBuffer fwdingmsg = new StringBuffer(500);
	fwdingmsg.append(" forwarding counters: ");
	for (int j = 0; j < forwardingClients.length; j++) {
	if( j!= 0 ) {
	fwdingmsg.append(", ");
	}
	fwdingmsg.append(forwardingClients[j].forwardCounter.get());
	}
	LOG.info(fwdingmsg.toString());

	// The test is just checking to make sure thaat the producer and consumer does not hang
	// due to the network hops take to route the message form the producer to the consumer.
	assertTrue("Recieved some messages since last sample", r>0);
	assertTrue("Produced some messages since last sample", p>0);

	}
	LOG.info("Sample done.");
	done.set(true);
	// Wait for the producer to finish.
	producer.join(1000*5);
	toConnection.close();
	fromConnection.close();

	}


	}