package org.opoo.samples.activemq;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicBoolean;

import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.Session;

import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.core.MessageCreator;

/**
 * JMS示例。
 * 
 * @author <a href="http://opoo.org">opoo.org</a>
 *
 */
public class ActiveMQJmsMain {
	private static final String QUEUE = "jmsLocalQueue";

	/**
	 * 消息接收（消费）线程。
	 */
	public static class ReceiveRunnable implements Runnable {
		private AtomicBoolean stopReceiving = new AtomicBoolean();
		private JmsTemplate consumerTemplate;

		public ReceiveRunnable(JmsTemplate consumerTemplate) {
			super();
			this.consumerTemplate = consumerTemplate;
		}

		public void run() {
			while (!stopReceiving.get()) {
				try {
					Message message = consumerTemplate.receive(QUEUE);
					System.out.println(Thread.currentThread() + " 收到消息：" + message);
				} catch (Exception e) {
					e.printStackTrace();
				}
			}
		}

		public void shutdown() {
			stopReceiving.set(true);
		}
	}

	/**
	 * 消息发送（生产）线程。
	 *
	 */
	public static class SendRunnable implements Runnable {
		private AtomicBoolean stopSending = new AtomicBoolean();
		private ArrayBlockingQueue<String> messagesToSend = new ArrayBlockingQueue<String>(1);
		private JmsTemplate producerTemplate;

		public SendRunnable(JmsTemplate producerTemplate) {
			super();
			this.producerTemplate = producerTemplate;
		}

		public void send(String message) {
			try {
				messagesToSend.put(message);
			} catch (InterruptedException e) {
				System.err.println("Unable to send even: " + e);
			}
		}

		public void run() {
			while (!stopSending.get()) {
				try {
					final String message = messagesToSend.take();
					producerTemplate.send(QUEUE, new MessageCreator() {
						public Message createMessage(Session session) throws JMSException {
							System.out.println(Thread.currentThread() + " 发出消息：" + message);
							return session.createTextMessage(message);
						}
					});
				} catch (Exception e) {
					e.printStackTrace();
				}
			}
		}

		public void shutdown() {
			stopSending.set(true);
		}
	}
	
	/**
	 * <p>mvn exec:java
	 * 
	 * 
	 * <p>1.在应用内部启动ActiveMQ服务：
	 * <p>直接执行 <code>mvn exec:java</code> 即可。
	 * 可通过系统属性 <code>msg.data.home.dir</code> 来指定JMS数据目录的位置
	 * <code>mvn exec:java -Dmsg.data.home.dir=/var/jmsDataDir</code> 或者 
	 * <code>mvn exec:java -Dmsg.data.home.dir=C:/jmsDataDir</code>。
	 * 
	 * <p>2. 独立启动ActiveMQ服务：
	 * <p>需要<a href="http://activemq.apache.org/download.html">下载ActiveMQ发行包</a>，并运行 bin/activemq 启动ActiveMQ服务。
	 * 然后通过以下命令运行本示例： <code>mvn exec:java -Dremote=true</code>
	 * 如果ActiveMQ服务和本示例的不在同一台机器上运行，则需要在运行时指定ActiveMQ的brokerURL
	 * <code>mvn exec:java -Dremote=true -Dbroker.url=tcp://192.168.1.99:61666</code>
	 * 
	 * @param args
	 */
	public static void main(String[] args){
		ClassPathXmlApplicationContext context;
		if("true".equalsIgnoreCase(System.getProperty("remote"))){
			context = new ClassPathXmlApplicationContext("spring-jms-remote.xml");
		}else{
			context = new ClassPathXmlApplicationContext("spring-jms.xml");
		}
		
    	JmsTemplate producerTemplate = (JmsTemplate) context.getBean("producerTemplate");
    	JmsTemplate consumerTemplate = (JmsTemplate) context.getBean("consumerTemplate");
    	
    	ReceiveRunnable receiveRunnable = new ReceiveRunnable(consumerTemplate);
    	SendRunnable sendRunnable = new SendRunnable(producerTemplate);
    	
    	ExecutorService executorService = Executors.newFixedThreadPool(2);
    	
    	executorService.execute(receiveRunnable);
    	executorService.execute(sendRunnable);
    	
    	while(true){
    		sendRunnable.send("Hello, 你好！" + System.currentTimeMillis());
    		try {
				Thread.sleep(1000L);//暂停1秒
			} catch (InterruptedException e) {
				//ignore
			}
    	}
	}
}