/***
    * Redistribution and use of this software and associated documentation
    * ("Software"), with or without modification, are permitted provided
    * that the following conditions are met:
    *
    * 1. Redistributions of source code must retain copyright
    *    statements and notices.  Redistributions must also contain a
    *    copy of this document.
    *
   * 2. Redistributions in binary form must reproduce the
   *    above copyright notice, this list of conditions and the
   *    following disclaimer in the documentation and/or other
   *    materials provided with the distribution.
   *
   * 3. The name "Exolab" must not be used to endorse or promote
   *    products derived from this Software without prior written
   *    permission of Exoffice Technologies.  For written permission,
   *    please contact info@exolab.org.
   *
   * 4. Products derived from this Software may not be called "Exolab"
   *    nor may "Exolab" appear in their names without prior written
   *    permission of Exoffice Technologies. Exolab is a registered
   *    trademark of Exoffice Technologies.
   *
   * 5. Due credit should be given to the Exolab Project
   *    (http://www.exolab.org/).
   *
   * THIS SOFTWARE IS PROVIDED BY EXOFFICE TECHNOLOGIES AND CONTRIBUTORS
   * ``AS IS'' AND ANY EXPRESSED OR IMPLIED WARRANTIES, INCLUDING, BUT
   * NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND
   * FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED.  IN NO EVENT SHALL
   * EXOFFICE TECHNOLOGIES OR ITS CONTRIBUTORS BE LIABLE FOR ANY DIRECT,
   * INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
   * (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
   * SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)
   * HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT,
   * STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
   * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED
   * OF THE POSSIBILITY OF SUCH DAMAGE.
   *
   * Copyright 2000-2004 (C) Exoffice Technologies Inc. All Rights Reserved.
   *
   * $Id: JmsSession.java,v 1.59 2004/01/21 04:08:13 tanderson Exp $
   */
  package org.exolab.jms.client;
  
  import java.io.Serializable;
  import java.util.Enumeration;
  import java.util.Hashtable;
  import java.util.Vector;
  
  import javax.jms.BytesMessage;
  import javax.jms.IllegalStateException;
  import javax.jms.JMSException;
  import javax.jms.MapMessage;
  import javax.jms.Message;
  import javax.jms.MessageListener;
  import javax.jms.ObjectMessage;
  import javax.jms.Session;
  import javax.jms.StreamMessage;
  import javax.jms.TextMessage;
  
  import org.apache.commons.logging.Log;
  import org.apache.commons.logging.LogFactory;
  
  import org.exolab.jms.message.BytesMessageImpl;
  import org.exolab.jms.message.MapMessageImpl;
  import org.exolab.jms.message.MessageConverter;
  import org.exolab.jms.message.MessageConverterFactory;
  import org.exolab.jms.message.MessageImpl;
  import org.exolab.jms.message.MessageSessionIfc;
  import org.exolab.jms.message.ObjectMessageImpl;
  import org.exolab.jms.message.StreamMessageImpl;
  import org.exolab.jms.message.TextMessageImpl;
  
  
  /***
   * This class implements a Session interface and supports a single threaded
   * context. A session supports multiple consumers and producers but only within
   * the context of a single thread.
   *
   * @version     $Revision: 1.59 $ $Date: 2004/01/21 04:08:13 $
   * @author      <a href="mailto:jima@exoffice.com">Jim Alateras</a>
   * @author      <a href="mailto:tma@netspace.net.au">Tim Anderson</a>
   */
  abstract class JmsSession
      implements Session, JmsMessageListener, MessageSessionIfc {
  
      /***
       * If true, indicates that the session has been closed
       */
      private volatile boolean _closed = false;
  
      /***
       * If true, indicates that the session is in the process of being closed
       */
      private volatile boolean _closing = false;
  
      /***
      * This flag determines whether message delivery is enabled or disabled.
      * Message delivery if disabled if the enclosing connection is stopped.
      */
     private volatile boolean _stopped = true;
 
     /***
      * If true, indicates that the session has been stopped. When started,
      * messages may be sent
      */
     private volatile boolean _started = false;
 
     /***
      * A transacted session is bounded by successive commit. If this variable
      * set to true then this session is transacted. This implies that the
      * session is always in a transaction and transactions are demarcated by
      * commit or rollback.
      */
     private final boolean _transacted;
 
     /***
      * Indicates whether the consumer or the client will
      * acknowledge any messages it receives. Ignored if the session is
      * transacted. Legal values are
      * <code>Session.AUTO_ACKNOWLEDGE</code>,
      * <code>Session.CLIENT_ACKNOWLEDGE</code> and
      * <code>Session.DUPS_OK_ACKNOWLEDGE</code>.
      */
     private final int _ackMode;
 
     /***
      * This is the owner of the session. It also maintains a remote stub to
      * the server which it can use for its remote work.
      */
     private JmsConnection _connection;
 
     /***
      * Maintains the a map of JmsMessageConsumer.getClientId() ->
      * JmsMessageConsumer objects
      */
     private Hashtable _consumers = new Hashtable();
 
     /***
      * Maintains a list of producers for the session
      */
     private Vector _producers = new Vector();
 
     /***
      * Maintain a collection of acked messages for this transacted
      * session. These messages are only sent to the server on commit.
      */
     private Vector _messagesToSend = new Vector();
 
     /***
      * The identifier of the session
      */
     private final String _sessionId;
 
     /***
      * Maintain a copy of the stub that connects this machine to the remote
      * server.
      */
     private JmsSessionStubIfc _stub = null;
 
     /***
      * This is the session's session listener which is used to receive all
      * messages associated with all consumers registered with this session.
      * Even if consumers have registered listeners messages for those consumers
      * will go to the session's message listener instead.
      */
     private MessageListener _listener = null;
 
     /***
      * The consumer Id seed to allocate to new consumers
      */
     private long _consumerIdSeed = 0;
 
     /***
      * Tracks the number of messages sent by this session
      */
     private long _publishCount;
 
     /***
      * The message cache holds all messages for the session, allocated by
      * a JmsConnectionConsumer.
      */
     private Vector _messageCache = new Vector();
 
     /***
      * Monitor used to block consumers, if the session has been
      * stopped, or no messages are available
      */
     private final Object _receiveLock = new Object();
 
     /***
      * The logger
      */
     private static final Log _log = LogFactory.getLog(JmsSession.class);
 
 
     /***
      * Construct a new <code>JmsSession</code>
      *
      * @param connection the owner of the session
      * @param transacted if <code>true</code>, the session is transacted.
      * @param ackMode indicates whether the consumer or the client will
      * acknowledge any messages it receives. This parameter will be ignored if
      * the session is transacted. Legal values are
      * <code>Session.AUTO_ACKNOWLEDGE</code>,
      * <code>Session.CLIENT_ACKNOWLEDGE</code> and
      * <code>Session.DUPS_OK_ACKNOWLEDGE</code>.
      * @throws JMSException if the session cannot be created
      */
     public JmsSession(JmsConnection connection, boolean transacted,
                       int ackMode) throws JMSException {
 
         if (connection == null) {
             throw new IllegalArgumentException(
                 "Argument 'connection' is null");
         }
 
         _connection = connection;
         _transacted = transacted;
         _ackMode = ackMode;
 
         // construct the remote stub
         _stub = connection.getJmsConnectionStub().createSession(
             _ackMode, transacted);
         _sessionId = _stub.getSessionId();
 
         // set up this instance to be a message listener
         _stub.setMessageListener(this);
 
         // now we need to check whether we should start the session
         if (!connection.isStopped()) {
             start();
         }
     }
 
     /***
      * Create a <code>BytesMessage</code>
      *
      * @return a new <code>BytesMessage</code>
      * @throws JMSException if the message can't be created
      */
     public BytesMessage createBytesMessage() throws JMSException {
         ensureOpen();
         return new BytesMessageImpl();
     }
 
     /***
      * Create a <code>MapMessage</code>
      *
      * @return a new <code>MapMessage</code>
      * @throws JMSException if the message can't be created
      */
     public MapMessage createMapMessage() throws JMSException {
         ensureOpen();
         return new MapMessageImpl();
     }
 
     /***
      * Create a <code>Message</code>
      *
      * @return a new <code>Message</code>
      * @throws JMSException if the message can't be created
      */
     public Message createMessage() throws JMSException {
         ensureOpen();
         return new MessageImpl();
     }
 
     /***
      * Create an <code>ObjectMessage</code>
      *
      * @return a new <code>ObjectMessage</code>
      * @throws JMSException if the message can't be created
      */
     public ObjectMessage createObjectMessage() throws JMSException {
         ensureOpen();
         return new ObjectMessageImpl();
     }
 
     /***
      * Create an <code>ObjectMessage</code>
      *
      * @param object the object to use to initialise the message
      * @return a new <code>ObjectMessage</code>
      * @throws JMSException if the message can't be created
      */
     public ObjectMessage createObjectMessage(Serializable object)
         throws JMSException {
         ensureOpen();
         ObjectMessageImpl result = new ObjectMessageImpl();
         result.setObject(object);
         return result;
     }
 
     /***
      * Create a <code>StreamMessage</code>
      *
      * @return a new <code>StreamMessage</code>
      * @throws JMSException if the message can't be created
      */
     public StreamMessage createStreamMessage() throws JMSException {
         ensureOpen();
         return new StreamMessageImpl();
     }
 
     /***
      * Create a <code>TextMessage</code>
      *
      * @return a new <code>TextMessage</code>
      * @throws JMSException if the message can't be created
      */
     public TextMessage createTextMessage() throws JMSException {
         ensureOpen();
         return new TextMessageImpl();
     }
 
     /***
      * Create a <code>TextMessage</code>
      *
      * @param text the text to use to initialise the message
      * @return a new <code>TextMessage</code>
      * @throws JMSException if the message can't be created
      */
     public TextMessage createTextMessage(String text) throws JMSException {
         ensureOpen();
         TextMessageImpl result = new TextMessageImpl();
         result.setText(text);
         return result;
     }
 
     /***
      * Determines if the session is transacted
      *
      * @return <code>true</code> if the session is transacted
      * @throws JMSException if the session is closed
      */
     public boolean getTransacted() throws JMSException {
         ensureOpen();
         return _transacted;
     }
 
     /***
      * Commit all messages done in this transaction
      *
      * @throws JMSException if the transaction cannot be committed
      */
     public synchronized void commit() throws JMSException {
         ensureOpen();
         ensureTransactional();
 
         // send all the cached messages to the server
         getJmsSessionStub().sendMessages(_messagesToSend);
         _publishCount += _messagesToSend.size();
         _messagesToSend.clear();
 
         // commit the session
         getJmsSessionStub().commit();
     }
 
     /***
      * Rollback any messages done in this transaction
      *
      * @throws JMSException if the transaction cannot be rolled back
      */
     public synchronized void rollback() throws JMSException {
         ensureOpen();
         ensureTransactional();
 
         // clear all the cached messages
         _messagesToSend.clear();
 
         // rollback the session
         getJmsSessionStub().rollback();
     }
 
     /***
      * Close the session. This call will block until a receive or message
      * listener in progress has completed. A blocked message consumer receive
      * call returns <code>null</code> when this session is closed.
      *
      * @throws JMSException if the session can't be closed
      */
     public synchronized void close() throws JMSException {
         if (!_closed) {
             _closing = true;
 
             // signal the stub that we are preparing to close the
             // connection.
             getJmsSessionStub().beforeClose();
 
             // must stop first before we close
             stop();
 
             // wake up any blocking consumers
             notifyConsumers();
 
             // go through all the producer and call close on them
             // respectively
             Enumeration producers = getProducers();
             while (producers.hasMoreElements()) {
                 JmsMessageProducer producer =
                     (JmsMessageProducer) producers.nextElement();
                 producer.close();
             }
 
             // go through all the consumer and call close on them
             // respectively
             Enumeration consumers = getConsumers();
             while (consumers.hasMoreElements()) {
                 JmsMessageConsumer consumer =
                     (JmsMessageConsumer) consumers.nextElement();
                 consumer.close();
             }
 
             // deregister this with the connection
             _connection.removeSession(this);
             _connection = null;
 
             // clear any cached messages or acks
             _messagesToSend.clear();
 
             // issue a close to the remote session. This will release any
             // allocated remote resources
             getJmsSessionStub().close();
             _stub = null;
 
             // update the session state
             _closed = true;
             _closing = false;
         }
     }
 
     /***
      * Stop message delivery in this session, and restart sending messages with
      * the oldest unacknowledged message
      *
      * @throws JMSException if the session can't be recovered
      */
     public synchronized void recover() throws JMSException {
         ensureOpen();
         if (!_transacted) {
             // let the server handle the recovery
             getJmsSessionStub().recover();
         } else {
             throw new IllegalStateException(
                 "Cannot recover from a transacted session");
         }
     }
 
     /***
      * Returns the message listener associated with the session
      *
      * @return the message listener associated with the session, or
      * <code>null</code> if no listener is registered
      * @throws JMSException if the session is closed
      */
     public MessageListener getMessageListener() throws JMSException {
         ensureOpen();
         return _listener;
     }
 
     /***
      * Sets the session's message listener.
      *
      * @param listener the session's message listener
      * @throws JMSException if the session is closed
      */
     public void setMessageListener(MessageListener listener)
         throws JMSException {
         ensureOpen();
         _listener = listener;
     }
 
     /***
      * Iterates through the list of messages added by an
      * {@link JmsConnectionConsumer}, sending them to the registered listener
      */
     public void run() {
         try {
             while (!_messageCache.isEmpty()) {
                 Message message = (Message) _messageCache.remove(0);
                 _listener.onMessage(message);
             }
         } catch (Exception exception) {
             _log.error("Error in the Session.run()", exception);
         } finally {
             // Clear message cache
             _messageCache.clear();
         }
     }
 
     /***
      * Set the message listener for a particular consumer.
      * <p>
      * If a listener is already registered for the consumer, it will be
      * automatically overwritten
      *
      * @param listener the message listener
      * @throws JMSException if the listener can't be set
      */
     public void setMessageListener(JmsMessageConsumer listener)
         throws JMSException {
         ensureOpen();
         enableAsynchronousDelivery(listener.getClientId(),
                                    listener.getLastMessageDelivered(), true);
     }
 
     /***
      * Remove a message listener
      *
      * @param listener the message listener to remove
      * @throws JMSException if the listener can't be removed
      */
     public void removeMessageListener(JmsMessageConsumer listener)
         throws JMSException {
 
         ensureOpen();
         enableAsynchronousDelivery(listener.getClientId(),
             listener.getLastMessageDelivered(), false);
     }
 
     /***
      * This will start message delivery to this session. If message delivery
      * has already started then this is a no-op.
      *
      * @throws JMSException if message delivery can't be started
      */
     public void start() throws JMSException {
         ensureOpen();
         if (_stopped) {
             getJmsSessionStub().startMessageDelivery();
             _stopped = false;
 
             // wake up any blocking consumers
             notifyConsumers();
         }
     }
 
     /***
      * This will stop message delivery to this session. If message delivery
      * has already stoped then this is a no-op.
      *
      * @throws JMSException if message delivery can't be stopped
      */
     public void stop() throws JMSException {
         ensureOpen();
         if (!_stopped) {
             getJmsSessionStub().stopMessageDelivery();
             _stopped = true;
 
             // wake up any blocking consumers
             notifyConsumers();
         }
     }
 
     /***
      * Acknowledge the specified message. This is only applicable for
      * CLIENT_ACKNOWLEDGE sessions. For other session types, the request
      * is ignored.
      * <p>
      * Acking a message automatically acks all those that have come
      * before it.
      *
      * @param message the message to acknowledge
      * @throws JMSException if the message can't be acknowledged
      */
     public void acknowledgeMessage(Message message) throws JMSException {
         ensureOpen();
         if (_ackMode == Session.CLIENT_ACKNOWLEDGE) {
             MessageImpl impl = (MessageImpl) message;
             getJmsSessionStub().acknowledgeMessage(impl.getClientId(),
                                                    impl.getAckMessageID());
         }
     }
 
     /***
      * Enable or disable asynchronous message delivery for the specified
      * client.
      *
      * @param clientId - the client identity
      * @param id - the last message delivered asynchronously
      * @param enable - <code>true</code> to enable; <code>false</code> to
      * disable
      * @throws JMSException if message delivery cannot be enabled or disabled
      */
     public void enableAsynchronousDelivery(long clientId, String id,
                                            boolean enable)
         throws JMSException {
 
         ensureOpen();
         getJmsSessionStub().enableAsynchronousDelivery(clientId, id, enable);
     }
 
     /***
      * Asynchronously deliver a message to a <code>MessageConsumer</code>
      *
      * @param message the message to deliver
      */
     public void onMessage(Message message) {
         if (message != null) {
             MessageImpl impl = (MessageImpl) message;
             impl.setJMSXRcvTimestamp(System.currentTimeMillis());
 
             // dispatch the message;
             execute(message);
         }
     }
 
     /***
      * Asynchronously deliver a set of message to a
      * <code>MessageConsumer</code>
      *
      * @param messages the messages to deliver
      */
     public void onMessages(Vector messages) {
         while (messages.size() > 0) {
             onMessage((Message) messages.remove(0));
         }
     }
 
     /***
      * Inform the session that there is a message available
      * for the message consumer with the specified identity
      *
      * @param clientId the identity of the client
      */
     public void onMessageAvailable(long clientId) {
         // wake up any blocking consumers
         notifyConsumers();
     }
 
     /***
      * This is the called to process messages asynchronously delivered by the
      * server. The session is then responsible for delivering it to the
      * appropriate registered consumer. If it cannot resolve the consumer then
      * it must log an exception
      * <p>
      * If the session has a registered listener then all messages will be
      * delivered to the session's listener instead of the individual consumer
      * message listeners.
      *
      * @param       object         received message
      */
     public synchronized void execute(Object object) {
         // if the session is closed then drop the object
         if (_closed) {
             _log.error("Received a message for a closed session");
             return;
         }
 
         MessageImpl message = (MessageImpl) object;
         long clientId = message.getClientId();
         JmsMessageConsumer consumer =
             (JmsMessageConsumer) _consumers.get(new Long(clientId));
 
         // tag the session that received this message
         message.setSession(this);
         if (consumer != null) {
             // if a listener is defined for the session then send all the
             // messages to that listener regardless if any consumers are
             // have registered listeners...bit confusing but this is what
             // I believe it should do
             if (_listener != null) {
                 _listener.onMessage(message);
             } else {
                 // send it to the appropriate consumer
                 consumer.onMessage(message);
             }
         } else {
             // consumer no longer active...so drop the message
             _log.error("Received a message for an inactive consumer");
         }
     }
 
     /***
      * Returns the session identifier
      *
      * @return the session identifier
      */
     public String getSessionId() {
         return _sessionId;
     }
 
     /***
      * Return the acknowledgement mode for the session
      *
      * @return the acknowledgement mode for the session
      */
     public int getAckMode() {
         return _ackMode;
     }
 
     /***
      * Fetch the next message for this client. If the session's ackMode is
      * client acknowledge then set the session for the message, othwerwise
      * ack the message before returning it.
      *
      * @param clientId the consumer identififer.
      * @param wait the maximum time to wait for a message, in milliseconds.
      * If <code>-1</code>, don't wait, if <code>0</code> wait indefinitely,
      * otherwise wait the specified time.
      * @return the received message, or <code>null</code>, if no message is
      * available
      * @throws JMSException if an error occurs retrieving the message
      */
     public Message retrieveMessage(long clientId, long wait)
         throws JMSException {
 
         ensureOpen();
 
         boolean breakOnNextRead = false;
         long start = System.currentTimeMillis();
         long end = start + wait;
         MessageImpl message = null;
         while (true) {
             synchronized (_receiveLock) {
                 if (_closing || _closed) {
                     // session is in the process of closing, or has been
                     // closed. Need to return null.
                     break;
                 } else if (_stopped) {
                     // connection has been stopped. No message can be returned,
                     // but receives continue to time out
                 } else {
                     // connection is started. Messages may be returned.
                     message = (MessageImpl) getJmsSessionStub().receiveMessage(
                         clientId, wait);
                 }
                 if (message != null) {
                     message.setSession(this);
                     break;
                 } else {
                     // if we have instructed to break, then exit the loop.
                     if (breakOnNextRead) {
                         break;
                     }
 
                     // no message was received. Block for the specified time
                     // until one of the following occurs:
                     // . a message is received
                     // . the receive times out
                     // . the session is closed
                     if (wait >= 0) {
                         try {
                             if (wait > 0) {
                                 // wait for a specific period of time
                                 _receiveLock.wait(wait);
                                 long current = System.currentTimeMillis();
                                 if (current >= end) {
                                     breakOnNextRead = true;
                                 } else {
                                     // update the time to wait. If the value
                                     // is zero then break on the next read
                                     wait = end - current;
                                     if (wait == 0) {
                                         breakOnNextRead = true;
                                     }
                                 }
                             } else {
                                 // wait indefinitely
                                 _receiveLock.wait();
                             }
                         } catch (InterruptedException ignore) {
                             // no-op
                         }
                     } else {
                         // exit the loop since the client is performing a non
                         // blocking read
                         break;
                     }
                 }
             }
         }
 
         return message;
     }
 
     /***
      * Fetch up to count messages from the endpoint. This should only
      * be called via a {@link JmsQueueBrowser}.
      *
      * @param clientId scoped to the session
      * @param count the max messages to retrieve.
      * @return the set of retrieve messages
      * @throws JMSException if messages can't be retrieved
      */
     public synchronized Vector retrieveMessages(long clientId, int count)
         throws JMSException {
         ensureOpen();
         return getJmsSessionStub().receiveMessages(clientId, count);
     }
 
     /***
      * Release local resources used by this session object
      *
      * @throws JMSException - if there is a problem completing this request
      */
     public void destroy() throws JMSException {
         if (!_closed) {
             _closing = true;
 
             // wake up any blocking consumers
             notifyConsumers();
 
             // go through all the producer and call close on them
             // respectively
             Enumeration producers = getProducers();
             while (producers.hasMoreElements()) {
                 JmsMessageProducer producer =
                     (JmsMessageProducer) producers.nextElement();
                 producer.destroy();
             }
 
             // go through all the consumer and call close on them
             // respectively
             Enumeration consumers = getConsumers();
             while (consumers.hasMoreElements()) {
                 JmsMessageConsumer consumer =
                     (JmsMessageConsumer) consumers.nextElement();
                 consumer.destroy();
             }
 
             // deregister this with the connection
             _connection.removeSession(this);
             _connection = null;
 
             // clear any cached messages or acks
             _messagesToSend.clear();
 
             // simply release the reference to the sever session
             _stub = null;
 
             // update the session state
             _closed = true;
             _closing = false;
         }
     }
 
     /***
      * Send the specified message to the server.
      *
      * @param message the message to send
      * @throws JMSException if the message can't be sent
      */
     protected synchronized void sendMessage(Message message)
         throws JMSException {
 
         if (_transacted) {
             // if the session is transacted then cache the message locally.
             // and wait for a commit or a rollback
             if (message instanceof MessageImpl) {
                 try {
                     message = (Message) ((MessageImpl) message).clone();
                 } catch (CloneNotSupportedException error) {
                     throw new JMSException(error.getMessage());
                 }
             } else {
                 message = convert(message);
             }
             _messagesToSend.addElement(message);
         } else {
             if (!(message instanceof MessageImpl)) {
                 message = convert(message);
             }
             getJmsSessionStub().sendMessage(message);
             _publishCount++;
         }
     }
 
     /***
      * Return an instance of the remote stub. This is set during object
      * creation time
      *
      * @return the remote stub
      */
     protected JmsSessionStubIfc getJmsSessionStub() {
         return _stub;
     }
 
     /***
      * Return a reference to the connection that created this session
      *
      * @return the owning connection
      */
     protected JmsConnection getConnection() {
         return _connection;
     }
 
     /***
      * This method checks the destination. If the destination is not temporary
      * then return true. If it is a temporary destination and it is owned by
      * this session's connection then it returns true. If it is a tmeporary
      * destination and it is owned by another connection then it returns false
      *
      * @param destination the destination to check
      * @return <code>true</code> if the destination is valid
      */
     protected boolean checkForValidTemporaryDestination(
         JmsDestination destination) {
         boolean result = false;
 
         if (destination.isTemporaryDestination()) {
             JmsTemporaryDestination temp =
                 (JmsTemporaryDestination) destination;
 
             // check  that this temp destination is owned by the session's
             // connection.
             if (temp.validForConnection(getConnection())) {
                 result = true;
             }
         } else {
             result = true;
         }
 
         return result;
     }
 
     /***
      * Returns a list of registered producers for the session
      *
      * @return an enumeration of the producers managed by the session
      */
     protected Enumeration getProducers() {
         return _producers.elements();
     }
 
     /***
      * Returns a list of registered consumers for the session
      *
      * @return an enumeration of the consumers managed by the session
      */
     protected Enumeration getConsumers() {
         return _consumers.elements();
     }
 
     /***
      * Returns the next seed value to be allocated to a new consumer
      *
      * @return  a unique identifier for a consumer for this session
      */
     protected long getNextConsumerId() {
         return ++_consumerIdSeed;
     }
 
     /***
      * Add a consumer to the list of consumers managed by this session
      *
      * @param consumer the consumer to add
      */
     protected void addConsumer(JmsMessageConsumer consumer) {
         _consumers.put(new Long(consumer.getClientId()), consumer);
     }
 
     /***
      * Remove the consumer with the specified id from the list of managed
      * consumers
      *
      * @param consumer the consumer to remove
      */
     protected void removeConsumer(JmsMessageConsumer consumer) {
         _consumers.remove(new Long(consumer.getClientId()));
     }
 
     /***
      * Add a producer to the list of producers managed by this session
      *
      * @param producer the producer to add
      */
     protected void addProducer(JmsMessageProducer producer) {
         _producers.addElement(producer);
     }
 
     /***
      * Remove the producer from the list of managed producers
      *
      * @param producer the producer to remove
      */
     protected void removeProducer(JmsMessageProducer producer) {
         _producers.remove(producer);
     }
 
     /***
      * Check if the session is closed
      *
      * @return <code>true</code> if the session is closed
      */
     protected final boolean isClosed() {
         return _closed;
     }
 
     /***
      * Add a message to the message cache. This message will be processed
      * when the run() method is called.
      *
      * @param message the message to add.
      */
     protected void addMessage(Message message) {
         _messageCache.addElement(message);
     }
 
     /***
      * Verifies that the session isn't closed
      *
      * @throws IllegalStateException if the session is closed
      */
     protected void ensureOpen() throws IllegalStateException {
         if (_closed) {
             throw new IllegalStateException(
                 "Cannot perform operation - session has been closed");
         }
     }
 
     /***
      * Verifies that the session is transactional
      *
      * @throws IllegalStateException if the session isn't transactional
      */
     private void ensureTransactional() throws IllegalStateException {
         if (!_transacted) {
             throw new IllegalStateException(
                 "Cannot perform operatiorn - session is not transactional");
         }
     }
 
     /***
      * Notifies any blocking synchronous consumers
      */
     private void notifyConsumers() {
         synchronized (_receiveLock) {
             _receiveLock.notifyAll();
         }
     }
 
     /***
      * Convert a message to its corresponding OpenJMS implementation
      *
      * @param message the message to convert
      * @return the OpenJMS implementation of the message
      * @throws JMSException for any error
      */
     private Message convert(Message message) throws JMSException {
         MessageConverter converter =
             MessageConverterFactory.create(message);
         return converter.convert(message);
     }
 
 }