package cn.amq.basis;
import java.io.IOException;
import java.util.List;
import java.util.TreeMap;

import javax.jms.Connection;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.Session;
import javax.jms.StreamMessage;
import javax.jms.Topic;
import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.transport.TransportListener;
import org.apache.log4j.Logger;
import com.jtiny.msgthread.JtinyMsgThread;
import com.jtiny.utility.JtinyEvent;
import com.jtiny.utility.Utils;
/*
 * 重载doSubscirbleJob
 * */
public class AmqSubscribleBasisJob extends JtinyMsgThread implements MessageListener{
	 static Logger logger = Logger.getLogger(AmqSubscribleBasisJob.class);
	 List<String> subscribleSubjectList;
	 private MessageConsumer[] messageConsumerArray=null;
	 private String amqServerUrl=null;
	 private TreeMap<String,Class > classTagNameMap=new TreeMap<String,Class>();
	 
	 private ActiveMQConnectionFactory factory=null;
	 private Connection connection;
	 private Session session;
	 
	 public AmqSubscribleBasisJob(String amqServerUrl,List<String> subscribleSubjectList){
		 super(50000);//防止amq消息到达过快，加大内部缓冲
		 this.subscribleSubjectList=subscribleSubjectList;
		 this.amqServerUrl=amqServerUrl;
		 messageConsumerArray=new MessageConsumer[subscribleSubjectList.size()];
	 }
	 public void init(){
		 try {
			 if(factory!=null)return;
			 factory = new ActiveMQConnectionFactory(amqServerUrl);
			 connection = factory.createConnection();
			 session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
			 for(int i=0;i<subscribleSubjectList.size();i++){
				Topic topic=session.createTopic(subscribleSubjectList.get(i));
				messageConsumerArray[i] = session.createConsumer(topic);
				messageConsumerArray[i].setMessageListener(this);
			 }
			((ActiveMQConnection) connection).addTransportListener(new MyTransportListen(this));
			 connection.start();
		 } catch (Exception e) {
			 	factory=null;
				logger.error(Utils.getStackTraceAsString(e));
			}
	 }
	final public  void handleEvent(JtinyEvent tinyEvent)
	{
		switch(tinyEvent.getMsgId())
		{
			case AmqConst.AMQ_SETUP_INIT_EVENT:{
				JtinyEvent msgEnityTimer=new JtinyEvent(AmqConst.AMQ_SETUP_INIT_EVENT,0,0,null,null,null);
				this.registerTimer(msgEnityTimer, 1000);
				if(tinyEvent.getIfParam()==AmqConst.AMQ_SOCK_ERROR_MSG)factory=null;
				this.init();
				break;
			}
			case AmqConst.AMQ_SUBSCRIBLE_MSG:{
				AmqBasisEvent amqBasisMsg=(AmqBasisEvent) tinyEvent.getObj();
				doSubscirbleJob(amqBasisMsg);
				break;
			}
			default:{
				logger.warn("amq received unknow message,the message type="+tinyEvent.getMsgId());
				break;
			}
		}
	}
	/*上层对此消息要重载实现*/
	public void doSubscirbleJob(AmqBasisEvent amqBasisMsg) {
		
		
	}
	final public void injectClassTagName(String tagName,Class<?> cl){
		classTagNameMap.put(tagName, cl);
	}
	public void onMessage(Message message) {
		if((message instanceof StreamMessage)==false)return;
		String msgType=null;
		String text=null;
		StreamMessage tMessage=(StreamMessage)message;
		try {
			byte[]container=new byte[1000*1024]; //原先100*1024大小时，若通道数过多会出现无法解析报文情况,因此消息不能大于此值.
			int len=tMessage.readBytes(container);
			text=new String(container,0,len);
			msgType=AmqBasisEvent.readMessageType(text);
			logger.debug("********************** subscrible text****************\r\n"+text);
			Class<?> cl=classTagNameMap.get(msgType);
			if(cl!=null){
				AmqBasisEvent amqBasisMsg=(AmqBasisEvent) cl.newInstance();
				amqBasisMsg.readTextToObject(text);
				JtinyEvent eventTimer=new JtinyEvent(AmqConst.AMQ_SUBSCRIBLE_MSG,0,0,null,null,amqBasisMsg);
				//统计收到多少个subscrible消息
				//BeanHelper.getSysStaticManager().plusRcvSubscribleEvent();
				if(this.getHighWater()>90){
					logger.error("onMessage: queue is full.msgType="+msgType);
				}else this.putEvent(eventTimer);
			}else{
				logger.warn("AmqSubscribleBasisJob: Received Unkonw Message! the text is:\r\n"+text);
			}
		} catch (Exception e) {
			logger.error("**********msgType="+msgType+";\r\n"+"text:"+text);
			logger.error(Utils.getStackTraceAsString(e));
		}
	}
}
class  MyTransportListen implements TransportListener{
	private AmqSubscribleBasisJob task=null;
	public MyTransportListen(AmqSubscribleBasisJob task){
		this.task=task;
	}
	public void onCommand(Object cmd) {  
	}
	public void onException(IOException arg0) {	
		JtinyEvent eventTimer=new JtinyEvent(AmqConst.AMQ_SETUP_INIT_EVENT,AmqConst.AMQ_SOCK_ERROR_MSG,0,null,null,null);
		this.task.putEvent(eventTimer);
	}
	public void transportInterupted() {		
	}
	public void transportResumed() {	
	}
	
}