epgdirsubscriber.java

基于JMS架构的电子节目单EPG下载子模块

package com.temobi.epg.publisher.bo;

import java.util.List;

import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.Session;
import javax.jms.TextMessage;
import javax.jms.Topic;
import javax.jms.TopicConnection;
import javax.jms.TopicConnectionFactory;
import javax.jms.TopicSession;
import javax.jms.TopicSubscriber;
import javax.naming.Context;
import javax.naming.InitialContext;
import javax.naming.NamingException;

import com.temobi.epg.publisher.EpgPublisherConstant;
import common.context.SystemContext;
import common.log.proxy.JLogger;
import common.log.proxy.LoggerFactory;


/**
 * EPG目录结构订阅者
 * 
 * @author lixu
 * 
 */
public class EpgDirSubscriber implements MessageListener {
	private static final JLogger log = LoggerFactory.getLogger(EpgDirSubscriber.class);
	
	private TopicConnection topicConnection;

	private TopicSession topicSession;

	private TopicSubscriber topicSubscriber;

	private Topic topic;

	public EpgDirSubscriber() throws JMSException, NamingException {
		SystemContext sysContext = SystemContext.getInstance();
		Context context = new InitialContext();
		TopicConnectionFactory topicFactory = (TopicConnectionFactory) context
				.lookup(sysContext.getValue(EpgPublisherConstant.FACTORY_JNDI));
		topicConnection = topicFactory.createTopicConnection();
		topicConnection.setExceptionListener(new JMSExceptionListener());
		topicSession = topicConnection.createTopicSession(false,
				Session.AUTO_ACKNOWLEDGE);

		topic = (Topic) context.lookup(sysContext
				.getValue(EpgPublisherConstant.TOPIC_JNDI));
		topicSubscriber = topicSession.createSubscriber(topic);
		topicSubscriber.setMessageListener(this);
		log.info("EpgDirSubscriber subscribed to topic: "
				+ sysContext.getValue(EpgPublisherConstant.TOPIC_JNDI));
		topicConnection.start();
	}

	public EpgDirSubscriber(String factoryJNDI, String topicJNDI, String user,
			String pwd) throws JMSException, NamingException {
		Context context = new InitialContext();
		TopicConnectionFactory topicFactory = (TopicConnectionFactory) context
				.lookup(factoryJNDI);
		topicConnection = topicFactory.createTopicConnection(user, pwd);
		topicSession = topicConnection.createTopicSession(false,
				Session.AUTO_ACKNOWLEDGE);
		topic = (Topic) context.lookup(topicJNDI);
		// topicSubscriber = topicSession.createDurableSubscriber(topic,
		// "jms-ex1dtpsd");
		topicSubscriber = topicSession.createSubscriber(topic);
		topicSubscriber.setMessageListener(this);
		log.info("EpgDirSubscriber subscribed to topic: " + topicJNDI);
		topicConnection.start();
	}

	public void onMessage(Message m) {
		try {
			String msg = ((TextMessage) m).getText();
			try {
				List<String> loadUrls = EpgDirTool.parseDirsFromXml(msg);
				EpgLoader loader = new EpgLoader(loadUrls);
				loader.load();
			} catch (Exception e) {
				log.error("load epg from remote host occur error:", e);
			}
		} catch (JMSException ex) {
			log.error(
					"Could not get text message: " + ex, ex);
		}
	}

	public void close() {
		try {
			topicConnection.setExceptionListener(null);
			topicSession.close();
			topicConnection.close();
		} catch (Exception e) {
		}
	}

}