messageserver.java

《j2ee开发全程实录》随书源码

package com.cownew.PIS.framework.server.msgMgr;

import java.util.ArrayList;
import java.util.List;
import java.util.Properties;

import javax.jms.DeliveryMode;
import javax.jms.JMSException;
import javax.jms.Queue;
import javax.jms.QueueConnection;
import javax.jms.QueueConnectionFactory;
import javax.jms.QueueReceiver;
import javax.jms.QueueSender;
import javax.jms.QueueSession;
import javax.jms.Session;
import javax.jms.TextMessage;
import javax.naming.Context;
import javax.naming.InitialContext;
import javax.naming.NamingException;

import org.apache.log4j.Logger;

import com.cownew.PIS.framework.common.msgMgr.MessageConstant;
import com.cownew.PIS.framework.common.msgMgr.MessageException;
import com.cownew.PIS.framework.server.helper.ServerConfig;
import com.cownew.PIS.framework.server.helper.ThreadVariableManager;
import com.cownew.PIS.framework.server.sessionMgr.ServerUserContext;
import com.cownew.ctk.common.ExceptionUtils;

public class MessageServer
{
	private QueueSession queueSession;

	private Queue queue;
	
	private QueueConnection queueConnection;

	public MessageServer() throws MessageException
	{
		super();

		Properties properties = new Properties();
		ServerConfig srvConfig = ServerConfig.getInstance();
		properties.put(Context.INITIAL_CONTEXT_FACTORY, srvConfig
				.getJMSContextFactory());
		properties.put(Context.PROVIDER_URL, srvConfig.getJMSUrl());
		try
		{
			Context context = new InitialContext(properties);
			// 获得JMS Topic连接队列工厂
			QueueConnectionFactory queueConnectionFactory = (QueueConnectionFactory) context
					.lookup(srvConfig.getJMSConnectionFactory());

			try
			{
				queueConnection = queueConnectionFactory
						.createQueueConnection();
				queueConnection.setClientID("abc");
				queue = (Queue) context
						.lookup(srvConfig.getJMSJndiName());
				queueSession = queueConnection.createQueueSession(false,
						Session.AUTO_ACKNOWLEDGE);
				
			} catch (JMSException e)
			{
				throw new MessageException(
						MessageException.MessageInitException,e);
			}

		} catch (NamingException e)
		{
			throw ExceptionUtils.toRuntimeException(e);
		}
	}

	public void sendMessage(String title, String message, String receiverId)
			throws MessageException
	{
		try
		{
			QueueSender queueSender = queueSession.createSender(queue);
			queueSender.setDeliveryMode(DeliveryMode.PERSISTENT);
			TextMessage txtMsg = queueSession.createTextMessage();
			txtMsg.setText(message);
			txtMsg.setStringProperty(MessageConstant.RECEIVERID, receiverId);
			txtMsg.setStringProperty(MessageConstant.TITLE, title);
			ThreadVariableManager varMgr = ThreadVariableManager.getInstance();
			ServerUserContext ctx = varMgr.getCurrentServerUserContext();
			String curUserId = ctx.getCurUserId();
			txtMsg.setStringProperty(MessageConstant.SENDERID, curUserId);
			queueSender.send(txtMsg);
		} catch (JMSException e)
		{
			throw new MessageException(MessageException.MessageSendException,e);
		}
	}
	
	public TextMessage[] receiveMessage(String userId)
	{
		try
		{
			QueueReceiver queueReceiver = queueSession.createReceiver(queue);// ,
			// MessageConstant.RECEIVERID + "='" + curUserId + "'");
			List listMsg = new ArrayList();

			while (true)
			{
				TextMessage message = (TextMessage) queueReceiver
						.receiveNoWait();
				if (message == null)
				{
					break;
				}
				listMsg.add(message);
			}
			return (TextMessage[]) listMsg.toArray(new TextMessage[listMsg
					.size()]);
		} catch (JMSException e)
		{
			throw new MessageException(MessageException.MessageRecvException,e);
		}
	}
	
	public TextMessage[] receiveMessage()
	{
		ThreadVariableManager varMgr = ThreadVariableManager.getInstance();
		ServerUserContext ctx = varMgr.getCurrentServerUserContext();
		String curUserId = ctx.getCurUserId();
		return receiveMessage(curUserId);
	}

	public void close()
	{
		try
		{
			if (queueSession != null)
				queueSession.close();
		} catch (Throwable e)
		{
			Logger.getLogger(getClass()).error(e, e);
		}

		try
		{
			if (queueConnection != null)
				queueConnection.close();
		} catch (Throwable e)
		{
			Logger.getLogger(getClass()).error(e, e);
		}
	}
}