/*
    * Copyright (C) The Apache Software Foundation. All rights reserved.
    *
    * This software is published under the terms of the Apache Software License
    * version 1.1, a copy of which has been included with this distribution in
    * the LICENSE file.
    *
    * $Id: MessengerSupport.java 155459 2005-02-26 13:24:44Z dirkv $
    */
  package org.apache.commons.messenger;
  
  import java.io.Serializable;
  import java.util.HashMap;
  import java.util.Map;
  
  import javax.jms.BytesMessage;
  import javax.jms.Connection;
  import javax.jms.ConnectionConsumer;
  import javax.jms.ConnectionFactory;
  import javax.jms.DeliveryMode;
  import javax.jms.Destination;
  import javax.jms.JMSException;
  import javax.jms.MapMessage;
  import javax.jms.Message;
  import javax.jms.MessageConsumer;
  import javax.jms.MessageListener;
  import javax.jms.MessageProducer;
  import javax.jms.ObjectMessage;
  import javax.jms.Queue;
  import javax.jms.QueueBrowser;
  import javax.jms.QueueConnection;
  import javax.jms.QueueSender;
  import javax.jms.QueueSession;
  import javax.jms.ServerSessionPool;
  import javax.jms.Session;
  import javax.jms.StreamMessage;
  import javax.jms.TextMessage;
  import javax.jms.Topic;
  import javax.jms.TopicConnection;
  import javax.jms.TopicPublisher;
  import javax.jms.TopicSession;
  
  import org.apache.commons.logging.Log;
  import org.apache.commons.logging.LogFactory;
  
  /** <p><code>MessengerSupport</code> is an abstract base class which implements
    * most of the functionality of Messenger. Derivations need to specify the
    * connection and session creation and the pooling strategy.</p>
    *
    * @author <a href="mailto:jstrachan@apache.org">James Strachan</a>
    * @version $Revision: 155459 $
    */
  public abstract class MessengerSupport implements Messenger {
  
      /** Logger */
      private static final Log log = LogFactory.getLog(MessengerSupport.class);
      private static final Log destinationLog = LogFactory.getLog("org.apache.commons.messenger.destination");
  
      private static final boolean CACHE_REQUESTOR = true;
  
      /** The name of the Messenger */
      private String name;
  
      /** 
       * Whether Queue's and Topic's are looked up using JNDI (true)
       * or wether they should be created on the fly 
       */
      private boolean jndiDestinations;
  
      /** are topic subscribers durable? */
      private boolean durable;
  
      /** the delivery mode used by default when sending messages */
      private int deliveryMode = Message.DEFAULT_DELIVERY_MODE;
  
      /** the durable name used for durable topic based subscriptions */
      private String durableName;
  
      /** 
       * whether local messages are ignored when topic based subscription is used
       * with a message selector 
       */
      private boolean noLocal;
  
      /** Should we cache the requestor object per thread? */
      private boolean cacheRequestors;
  
      /** A Map of ListenerKey objects to MessageConsumer objects */
      private Map listeners = new HashMap();
  
      /** whether MessageProducer instances should be cached or not */
      private boolean cacheProducers = true;
  
      public MessengerSupport() {
      }
  
      public String toString() {
          try {
              MessengerSession session = borrowMessengerSession();
             String answer = super.toString() + " session: " + session.toString();
             returnMessengerSession(session);
             return answer;
         }
         catch (Exception e) {
             return super.toString() + " session: " + e.toString();
         }
     }
 
     public Destination getDestination(String subject) throws JMSException {
         MessengerSession messengerSession = borrowMessengerSession();
         try {
             boolean debug = destinationLog.isInfoEnabled();
             Session session = messengerSession.getSession();
             if (messengerSession.isTopic()) {
                 if (debug) {
                     destinationLog.info("Using topic: " + subject);
                 }
                 return getTopic((TopicSession) session, subject);
             }
             else {
                 if (debug) {
                     destinationLog.info("Using queue: " + subject);
                 }
                 return getQueue((QueueSession) session, subject);
             }
         }
         finally {
             returnMessengerSession(messengerSession);
         }
     }
 
     public Destination createTemporaryDestination() throws JMSException {
         MessengerSession messengerSession = borrowMessengerSession();
         try {
             Session session = messengerSession.getSession();
             if (messengerSession.isTopic()) {
                 TopicSession topicSession = (TopicSession) session;
                 return topicSession.createTemporaryTopic();
             }
             else {
                 QueueSession queueSession = (QueueSession) session;
                 return queueSession.createTemporaryQueue();
             }
         }
         finally {
             returnMessengerSession(messengerSession);
         }
     }
 
     public void send(Destination destination, Message message) throws JMSException {
         MessengerSession session = borrowMessengerSession();
         try {
             MessageProducer producer = session.getMessageProducer(destination);
             if (session.isTopic()) {
                 ((TopicPublisher) producer).publish((Topic) destination, message);
             }
             else {
                 ((QueueSender) producer).send((Queue) destination, message);
             }
         }
         finally {
             returnMessengerSession(session);
         }
     }
 
     public Message receive(Destination destination) throws JMSException {
         MessengerSession session = borrowMessengerSession();
         MessageConsumer consumer = null;
         try {
             consumer = borrowMessageConsumer(session, session.getSession(), destination);
             return consumer.receive();
         }
         finally {
             returnMessageConsumer(consumer);
             returnMessengerSession(session);
         }
     }
 
     public Message receive(Destination destination, String selector) throws JMSException {
         MessengerSession session = borrowMessengerSession();
         MessageConsumer consumer = null;
         try {
             consumer = borrowMessageConsumer(session, session.getSession(), destination, selector);
             return consumer.receive();
         }
         finally {
             returnMessageConsumer(consumer);
             returnMessengerSession(session);
         }
     }
 
     public Message receive(Destination destination, long timeoutMillis) throws JMSException {
         MessengerSession session = borrowMessengerSession();
         MessageConsumer consumer = null;
         try {
             consumer = borrowMessageConsumer(session, session.getSession(), destination);
             return consumer.receive(timeoutMillis);
         }
         finally {
             returnMessageConsumer(consumer);
             returnMessengerSession(session);
         }
     }
 
     public Message receive(Destination destination, String selector, long timeoutMillis) throws JMSException {
         MessengerSession session = borrowMessengerSession();
         MessageConsumer consumer = null;
         try {
             consumer = borrowMessageConsumer(session, session.getSession(), destination, selector);
             return consumer.receive(timeoutMillis);
         }
         finally {
             returnMessageConsumer(consumer);
             returnMessengerSession(session);
         }
     }
 
     public Message receiveNoWait(Destination destination) throws JMSException {
         MessengerSession session = borrowMessengerSession();
         MessageConsumer consumer = null;
         try {
             consumer = borrowMessageConsumer(session, session.getSession(), destination);
             return consumer.receiveNoWait();
         }
         finally {
             returnMessageConsumer(consumer);
             returnMessengerSession(session);
         }
     }
 
     public Message receiveNoWait(Destination destination, String selector) throws JMSException {
         MessengerSession session = borrowMessengerSession();
         MessageConsumer consumer = null;
         try {
             consumer = borrowMessageConsumer(session, session.getSession(), destination, selector);
             return consumer.receiveNoWait();
         }
         finally {
             returnMessageConsumer(consumer);
             returnMessengerSession(session);
         }
     }
 
     public MessageConsumer createConsumer(Destination destination) throws JMSException {
         MessengerSession session = borrowMessengerSession();
         try {
             return createMessageConsumer(session, session.getSession(), destination);
         }
         finally {
             returnMessengerSession(session);
         }
     }
 
     public MessageConsumer createConsumer(Destination destination, String selector) throws JMSException {
         MessengerSession session = borrowMessengerSession();
         try {
             return createMessageConsumer(session, session.getSession(), destination, selector);
         }
         finally {
             returnMessengerSession(session);
         }
     }
 
     public void run() {
         // don't return sessions which throw an exception
         try {
             MessengerSession session = borrowMessengerSession();
             session.getSession().run();
             returnMessengerSession(session);
         }
         catch (JMSException e) {
             // ### ignore
         }
     }
 
     public ConnectionConsumer createConnectionConsumer(
         Destination destination,
         ServerSessionPool sessionPool,
         int maxMessages)
         throws JMSException {
         return createConnectionConsumer(destination, null, sessionPool, maxMessages);
     }
 
     public ConnectionConsumer createConnectionConsumer(
         Destination destination,
         String selector,
         ServerSessionPool sessionPool,
         int maxMessages)
         throws JMSException {
         Connection connection = getConnection();
         if (isTopic(connection)) {
             TopicConnection topicConnection = (TopicConnection) connection;
             if (isDurable()) {
                 return topicConnection.createDurableConnectionConsumer(
                     (Topic) destination,
                     getDurableName(),
                     selector,
                     sessionPool,
                     maxMessages);
             }
             else {
                 return topicConnection.createConnectionConsumer(
                     (Topic) destination,
                     selector,
                     sessionPool,
                     maxMessages);
             }
         }
         else {
             QueueConnection queueConnection = (QueueConnection) connection;
             return queueConnection.createConnectionConsumer((Queue) destination, selector, sessionPool, maxMessages);
         }
     }
 
     public abstract Connection getConnection() throws JMSException;
 
     // Listener API
     //-------------------------------------------------------------------------
     public void addListener(Destination destination, MessageListener listener) throws JMSException {
         if (listener instanceof MessengerListener) {
             MessengerListener messengerListener = (MessengerListener) listener;
             messengerListener.setMessenger(this);
         }
         MessengerSession session = borrowMessengerSession();
         try {
             MessageConsumer consumer = createMessageConsumer(session, session.getListenerSession(), destination);
             consumer.setMessageListener(listener);
             ListenerKey key = new ListenerKey(destination, listener);
             listeners.put(key, consumer);
         }
         finally {
             returnMessengerSession(session);
         }
     }
 
     public void addListener(Destination destination, String selector, MessageListener listener) throws JMSException {
 
         if (listener instanceof MessengerListener) {
             MessengerListener messengerListener = (MessengerListener) listener;
             messengerListener.setMessenger(this);
         }
         MessengerSession session = borrowMessengerSession();
         try {
             MessageConsumer consumer = createMessageConsumer(session, session.getListenerSession(), destination, selector);
             consumer.setMessageListener(listener);
             ListenerKey key = new ListenerKey(destination, listener, selector);
             listeners.put(key, consumer);
         }
         finally {
             returnMessengerSession(session);
         }
     }
 
     public void removeListener(Destination destination, MessageListener listener) throws JMSException {
         ListenerKey key = new ListenerKey(destination, listener);
         MessageConsumer consumer = (MessageConsumer) listeners.remove(key);
         if (consumer == null) {
             throw new JMSException("The given listener object has not been added for the given destination");
         }
         consumer.close();
     }
 
     public void removeListener(Destination destination, String selector, MessageListener listener)
         throws JMSException {
 
         ListenerKey key = new ListenerKey(destination, listener, selector);
         MessageConsumer consumer = (MessageConsumer) listeners.remove(key);
         if (consumer == null) {
             throw new JMSException("The given listener object has not been added for the given destination and selector");
         }
         consumer.close();
     }
 
     // Message factory methods
     //-------------------------------------------------------------------------
     public BytesMessage createBytesMessage() throws JMSException {
         MessengerSession session = borrowMessengerSession();
         try {
             return session.getSession().createBytesMessage();
         }
         finally {
             returnMessengerSession(session);
         }
     }
 
     public MapMessage createMapMessage() throws JMSException {
         MessengerSession session = borrowMessengerSession();
         try {
             return session.getSession().createMapMessage();
         }
         finally {
             returnMessengerSession(session);
         }
     }
 
     public Message createMessage() throws JMSException {
         MessengerSession session = borrowMessengerSession();
         try {
             return session.getSession().createMessage();
         }
         finally {
             returnMessengerSession(session);
         }
     }
 
     public ObjectMessage createObjectMessage() throws JMSException {
         MessengerSession session = borrowMessengerSession();
         try {
             return session.getSession().createObjectMessage();
         }
         finally {
             returnMessengerSession(session);
         }
     }
 
     public ObjectMessage createObjectMessage(Serializable object) throws JMSException {
         MessengerSession session = borrowMessengerSession();
         try {
             return session.getSession().createObjectMessage(object);
         }
         finally {
             returnMessengerSession(session);
         }
     }
 
     public StreamMessage createStreamMessage() throws JMSException {
         MessengerSession session = borrowMessengerSession();
         try {
             return session.getSession().createStreamMessage();
         }
         finally {
             returnMessengerSession(session);
         }
     }
 
     public TextMessage createTextMessage() throws JMSException {
         MessengerSession session = borrowMessengerSession();
         try {
             return session.getSession().createTextMessage();
         }
         finally {
             returnMessengerSession(session);
         }
     }
 
     public TextMessage createTextMessage(String text) throws JMSException {
         MessengerSession session = borrowMessengerSession();
         try {
             return session.getSession().createTextMessage(text);
         }
         finally {
             returnMessengerSession(session);
         }
     }
 
     public void commit() throws JMSException {
         MessengerSession session = borrowMessengerSession();
         try {
             session.getSession().commit();
         }
         finally {
             returnMessengerSession(session);
         }
     }
 
     public void rollback() throws JMSException {
         MessengerSession session = borrowMessengerSession();
         try {
             session.getSession().rollback();
         }
         finally {
             returnMessengerSession(session);
         }
     }
 
     public void close() throws JMSException {
         getSessionFactory().close();
     }
 
     /**
      * Creates a browser on the given Queue
      */
     public QueueBrowser createBrowser(Destination destination) throws JMSException {
         MessengerSession session = borrowMessengerSession();
         QueueBrowser browser = null;
         try {
             return createBrowser(session, destination);
         }
         finally {
             returnMessengerSession(session);
         }
     }
 
     /** Get the producer's default delivery mode. */
     public int getDeliveryMode(Destination destination) throws JMSException {
         MessengerSession session = borrowMessengerSession();
         int deliveryMode = 0;
         try {
             MessageProducer producer = session.getMessageProducer(destination);
             deliveryMode = producer.getDeliveryMode();
         }
         finally {
             returnMessengerSession(session);
         }
         return deliveryMode;
     }
 
     /** Set the producer's default delivery mode. */
     public void setDeliveryMode(Destination destination, int deliveryMode) throws JMSException {
         MessengerSession session = borrowMessengerSession();
         MessageProducer producer = null;
         try {
             producer = session.getMessageProducer(destination);
             producer.setDeliveryMode(deliveryMode);
         }
         finally {
             returnMessengerSession(session);
         }
     }
 
     /**  Get the producer's default priority. */
     public int getPriority(Destination destination) throws JMSException {
         MessengerSession session = borrowMessengerSession();
         MessageProducer producer = null;
         int priority = 0;
         try {
             producer = session.getMessageProducer(destination);
             priority = producer.getPriority();
         }
         finally {
 
             returnMessengerSession(session);
         }
         return priority;
     }
 
     /**   Set the producer's default priority. */
     public void setPriority(Destination destination, int priority) throws JMSException {
         MessengerSession session = borrowMessengerSession();
         MessageProducer producer = null;
         try {
             producer = session.getMessageProducer(destination);
             producer.setPriority(priority);
         }
         finally {
             returnMessengerSession(session);
         }
     }
 
     /**  Get the producer's default delivery mode. */
     public long getTimeToLive(Destination destination) throws JMSException {
         MessengerSession session = borrowMessengerSession();
         long timeToLive = 0;
         try {
             MessageProducer producer = session.getMessageProducer(destination);
             timeToLive = producer.getTimeToLive();
         }
         finally {
             returnMessengerSession(session);
         }
         return timeToLive;
     }
 
     /**  <p>Set the default length of time in milliseconds from its dispatch time that
      *   a produced message should be retained by the message system.</p>
      */
     public void setTimeToLive(Destination destination, long timeToLive) throws JMSException {
         MessengerSession session = borrowMessengerSession();
         try {
             MessageProducer producer = session.getMessageProducer(destination);
             producer.setTimeToLive(timeToLive);
         }
         finally {
             returnMessengerSession(session);
         }
     }
 
     /** Get an indication of whether message timestamps are disabled. */
     public boolean getDisableMessageTimestamp(Destination destination) throws JMSException {
         MessengerSession session = borrowMessengerSession();
         boolean value = false;
         try {
             MessageProducer producer = session.getMessageProducer(destination);
             value = producer.getDisableMessageTimestamp();
         }
         finally {
             returnMessengerSession(session);
         }
         return value;
     }
     
     /** Set whether message timestamps are disabled. */
     public void setDisableMessageTimestamp(Destination destination, boolean value) throws JMSException {
         MessengerSession session = borrowMessengerSession();
         try {
             MessageProducer producer = session.getMessageProducer(destination);
             producer.setDisableMessageTimestamp(value);
         }
         finally {
             returnMessengerSession(session);
         }
     }
 
     /** Extends the send capability to send by specifying additional options. */
     public void send(Destination destination, Message message, int deliveryMode, int priority, long timeToLive)
         throws JMSException {
         MessengerSession session = borrowMessengerSession();
         try {
             MessageProducer producer = session.getMessageProducer(destination);
             if (session.isTopic()) {
                 ((TopicPublisher) producer).publish((Topic) destination, message, deliveryMode, priority, timeToLive);
             }
             else {
                 ((QueueSender) producer).send((Queue) destination, message, deliveryMode, priority, timeToLive);
             }
         }
         finally {
             returnMessengerSession(session);
         }
     }
 
     /**  Get an indication of whether message IDs are disabled. */
     public boolean getDisableMessageID(Destination destination) throws JMSException {
         MessengerSession session = borrowMessengerSession();
         boolean value = false;
         try {
             MessageProducer producer = session.getMessageProducer(destination);
             value = producer.getDisableMessageID();
         }
         finally {
             returnMessengerSession(session);
         }
         return value;
     }
 
     /** Set whether message IDs are disabled. */
     public void setDisableMessageID(Destination destination, boolean value) throws JMSException {
         MessengerSession session = borrowMessengerSession();
         try {
             MessageProducer producer = session.getMessageProducer(destination);
             producer.setDisableMessageID(value);
         }
         finally {
             returnMessengerSession(session);
         }
     }
 
     // Properties
     //-------------------------------------------------------------------------
     /** Gets the name that this Messenger is called in a MessengerManager */
     public String getName() {
         return name;
     }
 
     /** Sets the name that this Messenger is called in a MessengerManager */
     public void setName(String name) {
         this.name = name;
     }
 
     /** Setter for jndiDestinations */
     public void setJndiDestinations(boolean jndiDestinations) {
         this.jndiDestinations = jndiDestinations;
     }
 
     /** Getter for jndiDestinations */
     public boolean isJndiDestinations() {
         return jndiDestinations;
     }
 
     /** Gets whether topic subscribers are durable or not */
     public boolean isDurable() {
         return durable;
     }
 
     /** Sets whether topic subscribers are durable or not */
     public void setDurable(boolean durable) {
         this.durable = durable;
     }
 
     /** Gets whether we should cache the requestor object per thread? */
     public boolean isCacheRequestors() {
         return cacheRequestors;
     }
 
     /** Sets whether we should cache the requestor object per thread? */
     public void setCacheRequestors(boolean cacheRequestors) {
         this.cacheRequestors = cacheRequestors;
     }
 
     /** @return the durable name used for durable topic based subscriptions */
     public String getDurableName() {
         return durableName;
     }
 
     /** Sets the durable name used for durable topic based subscriptions */
     public void setDurableName(String durableName) {
         this.durableName = durableName;
     }
 
     /** 
      * Gets whether local messages are ignored when topic based subscription is used
      * with a message selector 
      */
     public boolean isNoLocal() {
         return noLocal;
     }
 
     /** 
      * Sets whether local messages are ignored when topic based subscription is used
      * with a message selector 
      */
     public void setNoLocal(boolean noLocal) {
         this.noLocal = noLocal;
     }
 
     /** Gets whether MessageProducer instances should be cached or not, which defaults to true */
     public boolean isCacheProducers() {
         return cacheProducers;
     }
 
     /** Sets whether MessageProducer instances should be cached or not, which defaults to true */
     public void setCacheProducers(boolean cacheProducers) {
         this.cacheProducers = cacheProducers;
     }
 
     /**
      * Returns the delivery mode used on messages sent via this Messenger
      * @return int
      */
     public int getDeliveryMode() {
         return deliveryMode;
     }
 
     /**
      * Sets the delivery mode used on messages sent via this Messenger
      * @param deliveryMode The deliveryMode to set
      */
     public void setDeliveryMode(int deliveryMode) {
         this.deliveryMode = deliveryMode;
     }
 
     /**
      * Sets whether message delivery should be persistent or not
      * 
      * @param persistentDelivery
      */
     public void setPersistentDelivery(boolean persistentDelivery) {
         if (persistentDelivery) {
             setDeliveryMode(DeliveryMode.PERSISTENT);
         }
         else {
             setDeliveryMode(DeliveryMode.NON_PERSISTENT);
         }
     }
 
     // Implementation methods
     //-------------------------------------------------------------------------
 
     protected abstract MessengerSession borrowMessengerSession() throws JMSException;
 
     protected abstract void returnMessengerSession(MessengerSession session);
 
     protected abstract boolean isTopic(Connection connection) throws JMSException;
 
     protected abstract boolean isTopic(ConnectionFactory factory) throws JMSException;
 
     /** @return a newly created message producer for the given session and destination */
     protected MessageProducer createMessageProducer(MessengerSession messengerSession, Destination destination) throws JMSException {
 
         MessageProducer answer = null;
         Session session = messengerSession.getSession();
         if (messengerSession.isTopic()) {
             TopicSession topicSession = (TopicSession) session;
             answer = topicSession.createPublisher((Topic) destination);
         }
         else {
             QueueSession queueSession = (QueueSession) session;
             answer = queueSession.createSender((Queue) destination);
         }
 
         // configure the MessageProducer
         if (deliveryMode != Message.DEFAULT_DELIVERY_MODE) {
             answer.setDeliveryMode(deliveryMode);
         }
         return answer;
     }
 
     /** @return a MessageConsumer for the given session and destination */
     protected MessageConsumer borrowMessageConsumer(MessengerSession messengerSession, Session session, Destination destination)
         throws JMSException {
 
         MessageConsumer consumer = createMessageConsumer(messengerSession, session, destination);
 
         if (log.isDebugEnabled()) {
             log.debug("Created new consumer: " + consumer + " on destination: " + destination);
         }
 
         return consumer;
     }
 
     /** @return a MessageConsumer for the given session, destination and selector */
     protected MessageConsumer borrowMessageConsumer(MessengerSession messengerSession, Session session, Destination destination, String selector)
         throws JMSException {
 
         MessageConsumer consumer = createMessageConsumer(messengerSession, session, destination, selector);
 
         if (log.isDebugEnabled()) {
             log.debug(
                 "Created new consumer: " + consumer + " on destination: " + destination + " selector: " + selector);
         }
 
         return consumer;
     }
 
     /** 
      * Returns a message consumer back to the pool. 
      * By default this method will close message consumers though we should
      * be able to cache then
      */
     protected void returnMessageConsumer(MessageConsumer messageConsumer) throws JMSException {
         if (log.isDebugEnabled()) {
             log.debug("Closing consumer: " + messageConsumer);
         }
 
         if (messageConsumer != null) {
             messageConsumer.close();
         }
     }
 
     /** @return a new MessageConsumer for the given session and destination */
     protected MessageConsumer createMessageConsumer(MessengerSession messengerSession, Session session, Destination destination)
         throws JMSException {
         if (messengerSession.isTopic()) {
             TopicSession topicSession = (TopicSession) session;
             if (isDurable()) {
                 return topicSession.createDurableSubscriber((Topic) destination, getDurableName(), null, isNoLocal());
             }
             else {
                 return topicSession.createSubscriber((Topic) destination, null, isNoLocal());
             }
         }
         else {
             QueueSession queueSession = (QueueSession) session;
             return queueSession.createReceiver((Queue) destination);
         }
     }
 
     /** @return a new MessageConsumer for the given session, destination and selector */
     protected MessageConsumer createMessageConsumer(
         MessengerSession messengerSession,
         Session session,
         Destination destination,
         String selector)
         throws JMSException {
         if (messengerSession.isTopic()) {
             TopicSession topicSession = (TopicSession) session;
             if (isDurable()) {
                 return topicSession.createDurableSubscriber(
                     (Topic) destination,
                     getDurableName(),
                     selector,
                     isNoLocal());
             }
             else {
                 return topicSession.createSubscriber((Topic) destination, selector, isNoLocal());
             }
         }
         else {
             QueueSession queueSession = (QueueSession) session;
             return queueSession.createReceiver((Queue) destination, selector);
         }
     }
 
     /** @return a new QueueBrowser for the given session and destination */
     protected QueueBrowser createBrowser(MessengerSession session, Destination destination) throws JMSException {
         if (session.isTopic()) {
             return null;
         }
         else {
             QueueSession queueSession = (QueueSession) session.getSession();
             return queueSession.createBrowser((Queue) destination);
         }
     }
 
     protected Queue getQueue(QueueSession session, String subject) throws JMSException {
         // XXXX: might want to cache
         return session.createQueue(subject);
     }
 
     protected Topic getTopic(TopicSession session, String subject) throws JMSException {
         // XXXX: might want to cache
         return session.createTopic(subject);
     }
 }