import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.DeliveryMode;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.MessageConsumer;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;

import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.ScheduledMessage;
import org.apache.activemq.command.ActiveMQDestination;
 
/**
 * <b>function:</b> 訊息傳送者
 * @author hoojo
 * @createDate 2013-6-19 上午11:26:43
 * @file MessageSender.java
 * @package com.hoo.mq.jms
 * @project ActiveMQ-5.8
 * @blog http://blog.csdn.net/IBM_hoojo
 * @email 
 
[email protected]
 * @version 1.0
 */
public class TestActiveMQ 
{ 
    // tcp 地址
    public static final String BROKER_URL = "failover:(tcp://localhost:61616)";
    // 目標，在ActiveMQ管理員控制檯建立 http://localhost:8161/admin/queues.jsp
    public static final String DESTINATION = "kagula.queue";
    
    /**
     * <b>function:</b> 傳送訊息
     * @author hoojo
     * @createDate 2013-6-19 下午12:05:42
     * @param session
     * @param producer
     * @throws Exception
     */    
    public static void sendMessage(Session session, MessageProducer producer) throws Exception {
    	
    	//例一：
    	String message = "直接傳送資料";
    	TextMessage tm = session.createTextMessage(message);
    	producer.send(tm);

    	//例二：
        //需要修改activemq.xml才能生效，在<broker>裡新增屬性schedulerSupport="true"
    	message = "延時10秒傳送資料";
    	TextMessage tm2 = session.createTextMessage("Send Message After 10 seconds!"); 
    	long delayTime = 1 * 10 * 1000;    	
    	tm2.setLongProperty(ScheduledMessage.AMQ_SCHEDULED_DELAY, delayTime); 
    	producer.send(tm2);
    }
    
    public static void run() throws Exception {
        
        Connection connection = null;
        Session session = null;
        try {
            // 建立連結工廠
            ConnectionFactory factory = new ActiveMQConnectionFactory(
            		ActiveMQConnection.DEFAULT_USER, 
            		ActiveMQConnection.DEFAULT_PASSWORD, BROKER_URL);
            // 通過工廠建立一個連線
            connection = factory.createConnection();
            // 啟動連線
            connection.start();
            // 建立一個session會話
            session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);
            // 建立一個訊息佇列
            Destination destination = session.createQueue(DESTINATION);
            // 建立訊息製作者
            MessageProducer producer = session.createProducer(destination);
            
            // 設定持久化模式
            producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
            sendMessage(session, producer);
            // 提交會話
            session.commit();
            
        } catch (Exception e) {
            throw e;
        } finally {
            // 關閉釋放資源
            if (session != null) {
                session.close();
            }
            if (connection != null) {
                connection.close();
            }
        }
    }
    
    public static void ClearMessage()
    {
    	// ConnectionFactory ：連線工廠，JMS 用它建立連線
        ConnectionFactory connectionFactory;
        // Connection ：JMS 客戶端到JMS Provider 的連線
        Connection connection = null;
        // Session： 一個傳送或接收訊息的執行緒
        Session session;
        // Destination ：訊息的目的地;訊息傳送給誰.
        Destination destination;
        // 消費者，訊息接收者
        MessageConsumer consumer;
        connectionFactory = new ActiveMQConnectionFactory(
                ActiveMQConnection.DEFAULT_USER,
                ActiveMQConnection.DEFAULT_PASSWORD,
                BROKER_URL);
        try {
            // 構造從工廠得到連線物件
            connection = connectionFactory.createConnection();
            // 啟動
            connection.start();
            // 獲取操作連線
            session = connection.createSession(Boolean.FALSE,
                    Session.AUTO_ACKNOWLEDGE);
            // 獲取session注意引數值xingbo.xu-queue是一個伺服器的queue，須在在ActiveMq的console配置
            destination = session.createQueue(DESTINATION);
            consumer = session.createConsumer(destination);
            while (true) {
                TextMessage message = (TextMessage) consumer.receive(10);
                if (null != message) {
                    System.out.println("把訊息拿下來但不處理！");
                } else {
                    break;
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            try {
                if (null != connection)
                    connection.close();
            } catch (Throwable ignore) {
            }
        }
    }
    
    /**
     * 刪除佇列
     * 如果有consumer連在佇列上，刪除佇列會失敗！
     * 所以不建議使用
     * 這個函式有程序阻塞問題
     * @param url
     * @param queueName
     */
    public static void ClearQueue(String url,String queueName)
    {
    	ActiveMQConnection con = null;
    	try {
    		con = (ActiveMQConnection) new ActiveMQConnectionFactory(url).createConnection();   	
    		Destination queue=null;
    		queue = con.createSession(false, Session.AUTO_ACKNOWLEDGE).createQueue(queueName);
   			con.start();
   			con.destroyDestination((ActiveMQDestination) queue);
    	} catch (JMSException e) 
    	{
    		e.printStackTrace();
    	} finally {
    		try
    		{
    			con.stop();
    		} catch(Exception e)
    		{
    			e.printStackTrace();
    		}
    	}//end finally
    }//end func
    
    /*
     * Active MQ後臺管理頁面，使用者名稱密碼分別為admin,admin
     * http://127.0.0.1:8161/admin/
     * */
    public static void main(String[] args) throws Exception {
    	System.out.println("program begin!");
    	
    	//刪除佇列
    	//ClearQueue(BROKER_URL,DESTINATION);
    	
    	//通過取佇列中的訊息，把佇列中的訊息刪除
    	ClearMessage();
    	
    	//傳送訊息示例
    	TestActiveMQ.run();
    	
    	System.out.println("program end!");
    }//end function
}//end class