/*
 * JBoss, Home of Professional Open Source.
 * Copyright 2009, Red Hat Middleware LLC, and individual contributors
 * as indicated by the @author tags. See the copyright.txt file in the
 * distribution for a full listing of individual contributors.
 *
 * This is free software; you can redistribute it and/or modify it
 * under the terms of the GNU Lesser General Public License as
 * published by the Free Software Foundation; either version 2.1 of
 * the License, or (at your option) any later version.
 *
 * This software is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
 * Lesser General Public License for more details.
 *
 * You should have received a copy of the GNU Lesser General Public
 * License along with this software; if not, write to the Free
 * Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA
 * 02110-1301 USA, or see the FSF site: http://www.fsf.org.
 */

import java.util.Properties;
import java.util.concurrent.Semaphore;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.logging.Level;
import java.util.logging.Logger;

import javax.jms.BytesMessage;
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.DeliveryMode;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.Topic;
import javax.jms.TopicSubscriber;
import javax.naming.Context;
import javax.naming.InitialContext;
import javax.naming.NamingException;

public class ProducerWithGap {

	// **********************************************************************************************************
	// parameters VVVVV
	static final String JNDI_ADDRESS = "jnp://127.0.0.1:1099,jnp://127.0.0.2:1099";

	static final String CF_NAME = "cn=PUT_TEST.Extended.QueueConnectionFactory";

	static final String TOPIC_NAME = "cn=PUT_TEST2.SECREF.NOTIFY.SECRATG.UPDATE.BRIDGE";

	static int TOTAL_MSGS = 2000000;

	static int MSG_SIZE = 10 * 1024;

	// When we should start consuming the first consumer
	static final int GAP_BETWEEN_CONSUMERS = 20000;

	// parameters ^^^^^^^^^
	// **********************************************************************************************************

	static final String propertyNameForCounter = "msgIndex";

	// latch for the first consumer
	static Semaphore firstSemaphore = new Semaphore(0);
	static Semaphore secondSemaphore = new Semaphore(0);

	static Topic topic;

	static AtomicInteger errors = new AtomicInteger(0);

	static class LoadProducer extends Thread {
		final ConnectionFactory cf;

		final Semaphore releaseSemaphore;

		final int releaseAt;

		LoadProducer(ConnectionFactory cf, Semaphore releaseSemaphore,
				int releaseAt) throws Exception {
			this.cf = cf;
			this.releaseAt = releaseAt;
			this.releaseSemaphore = releaseSemaphore;
		}

		@Override
		public void run() {

			Connection connection = null;

			Session session = null;

			MessageProducer prod = null;

			try {

				System.out.println("Starting producer for " + topic);

				connection = cf.createConnection();

				session = connection.createSession(true,
						Session.SESSION_TRANSACTED);

				prod = session.createProducer(topic);
				prod.setDeliveryMode(DeliveryMode.PERSISTENT);
				prod.setTimeToLive(120000);

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

					if (i >= releaseAt && releaseSemaphore != null) {
						releaseSemaphore.release();
					}

					BytesMessage msg = session.createBytesMessage();
					msg.setIntProperty(propertyNameForCounter, i);
					msg.writeBytes(new byte[MSG_SIZE]);
					prod.send(msg);

					if ((i + 1) % 10 == 0) {
						session.commit();
					}

					if (i % 1000 == 0) {
						System.out.println("Address " + topic + " had " + i
								+ " messages sent");
					}
				}
			} catch (Exception e) {
				errors.incrementAndGet();
				e.printStackTrace();
			} finally {
				if (releaseSemaphore != null) {
					releaseSemaphore.release(releaseAt);
				}

				try {
					session.commit();
					connection.close();
				} catch (Exception e) {
					e.printStackTrace();
					// don't care
				}
			}
		}
	}

	static class LoadConsumer extends Thread {
		final ConnectionFactory cf;

		final String connectionIDName;

		final String subsName;

		final Semaphore releaseLatch;

		// We need a gap between consumers
		final Semaphore releaseGapSemaphore;

		final int releaseGap;

		LoadConsumer(ConnectionFactory cf, String connectionIDName,
				String subsName, Semaphore releaseLatch,
				Semaphore releaseGapSemaphore, int releaseGap) throws Exception {
			this.cf = cf;
			this.subsName = subsName;
			this.connectionIDName = connectionIDName;
			this.releaseLatch = releaseLatch;
			this.releaseGapSemaphore = releaseGapSemaphore;
			this.releaseGap = releaseGap;
		}

		@Override
		public void run() {

			Session session = null;

			Connection connection = null;

			TopicSubscriber subs = null;
			while (true) {

				try {

					System.out.println("Created consumer for " + topic + "/"
							+ connectionIDName + "/" + subsName);

					connection = cf.createConnection();
					connection.setClientID(connectionIDName);
					connection.start();
					session = connection.createSession(true,
							Session.SESSION_TRANSACTED);
					subs = session.createDurableSubscriber(topic, subsName);

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

						releaseLatch.acquire();

						if (i == 0) {
							System.out.println("Starting consuming on subs="
									+ subsName);
						}

						if (i >= releaseGap && releaseGapSemaphore != null) {
							releaseGapSemaphore.release();
						}

						BytesMessage msg = (BytesMessage) subs.receive(60000);

						if (msg == null) {
							errors.incrementAndGet();
							System.out
									.println("Error: Couldn't receive a message in 60 seconds");
							break;
						}

						// if (i != msg.getIntProperty(propertyNameForCounter))
						// {
						// System.out.println("Message received " +
						// msg.getIntProperty(propertyNameForCounter) +
						// " was received out of sequence. It should be " + i +
						// " at " + this.subsName);
						// }

						session.commit();

						if (i % 10 == 0) {
							System.out.println("Subscription " + this.subsName
									+ " received " + i + " messages");
						}

						if (i % 1000 == 0) {
							System.out.println("Subscription " + this.subsName
									+ " received " + i + " messages");
						}
					}
					
					System.out.println("Finishing consumption on " + this.subsName);
					return;
				} catch (Exception e) {
					errors.incrementAndGet();
					e.printStackTrace();
				} finally {
					if (releaseGapSemaphore != null) {
						releaseGapSemaphore.release(releaseGap);
					}

					try {
						session.commit();
						session.close();
					} catch (Exception e) {
						e.printStackTrace();
						// don't care
					}
				}
			}
		}
	}

	/**
	 * @param args
	 *            the command line arguments
	 */
	public static void main(String[] args) {
		Context context = null;

		try {

			Properties properties = new Properties();
			properties.setProperty("java.naming.factory.initial",
					"org.jnp.interfaces.NamingContextFactory");
			properties.setProperty("java.naming.provider.url", JNDI_ADDRESS);
			properties.setProperty("java.naming.factory.url.pkgs",
					"org.jboss.naming:org.jnp.interfaces");

			context = new InitialContext(properties);

			ConnectionFactory cf = (ConnectionFactory) context.lookup(CF_NAME);

			topic = (Topic) context.lookup(TOPIC_NAME);

			LoadProducer prod = new LoadProducer(cf, firstSemaphore,
					GAP_BETWEEN_CONSUMERS);
			LoadConsumer cons1 = new LoadConsumer(cf, "cn1", "sub1",
					firstSemaphore, secondSemaphore, GAP_BETWEEN_CONSUMERS);
			LoadConsumer cons2 = new LoadConsumer(cf, "cn2", "sub2",
					secondSemaphore, null, 0);

			cons1.start();
			cons2.start();
			prod.start();

			prod.join();
			cons2.join();
			cons1.join();

			if (errors.get() != 0) {
				System.out.println("We had errors reported on the threads!!");
				System.out.println("FAILURE!!!!");
				System.exit(-1);
			}

		} catch (Exception e) {
			Logger.getLogger(ConsumerApp.class.getName()).log(Level.SEVERE,
					null, e);
		} finally {
			if (context != null) {
				try {
					context.close();
				} catch (NamingException ex) {
					Logger.getLogger(ConsumerApp.class.getName()).log(
							Level.SEVERE, null, ex);
				}
			}
		}
	}

}