/***
    * 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-2003 (C) Exoffice Technologies Inc. All Rights Reserved.
   *
   * $Id: JmsServerSession.java,v 1.89 2003/08/17 01:32:26 tanderson Exp $
   *
   * Date         Author  Changes
   * 04/07/2000   jima    Created
   * 04/25/2000   jima    Changes to the interface
   */
  package org.exolab.jms.server;
  
  import java.util.HashMap;
  import java.util.Iterator;
  import java.util.Vector;
  
  import javax.jms.DeliveryMode;
  import javax.jms.Destination;
  import javax.jms.InvalidDestinationException;
  import javax.jms.InvalidSelectorException;
  import javax.jms.JMSException;
  import javax.jms.Message;
  import javax.jms.MessageConsumer;
  import javax.jms.Session;
  import javax.transaction.TransactionManager;
  import javax.transaction.xa.XAException;
  import javax.transaction.xa.XAResource;
  import javax.transaction.xa.Xid;
  
  import org.apache.commons.logging.Log;
  import org.apache.commons.logging.LogFactory;
  
  import org.exolab.jms.JMSErrorCodes;
  import org.exolab.jms.client.JmsMessageListener;
  import org.exolab.jms.client.JmsQueue;
  import org.exolab.jms.client.JmsTopic;
  import org.exolab.jms.message.MessageHandle;
  import org.exolab.jms.message.MessageId;
  import org.exolab.jms.message.MessageImpl;
  import org.exolab.jms.messagemgr.ConsumerEndpoint;
  import org.exolab.jms.messagemgr.ConsumerManager;
  import org.exolab.jms.messagemgr.DestinationManager;
  import org.exolab.jms.messagemgr.InternalMessageListener;
  import org.exolab.jms.messagemgr.MessageMgr;
  import org.exolab.jms.messagemgr.QueueBrowserEndpoint;
  import org.exolab.jms.messagemgr.ResourceManager;
  import org.exolab.jms.messagemgr.ResourceManagerException;
  
  
  /***
   * A session represents a server side endpoint to the JMSServer. A client can
   * create producers, consumers and destinations through the session in addi-
   * tion to other functions. A session has a unique identifer which is a comb-
   * ination of clientId-connectionId-sessionId.
   * <p>
   * A session represents a single-threaded context which implies that it cannot
   * be used with more than one thread concurrently. Threads registered with this
   * session are synchronized.
   * <p>
   * Finally, instances of this object can only be created by classes within the
   * same package.
  *
  * @version     $Revision: 1.89 $ $Date: 2003/08/17 01:32:26 $
  * @author      <a href="mailto:jima@exoffice.com">Jim Alateras</a>
  * @author      <a href="mailto:tma@netspace.net.au">Tim Anderson</a>
  * @see         JmsServerConnection
  */
 public class JmsServerSession
     implements InternalMessageListener, XAResource {
 
     /***
      * The client identifies the owner of the session.
      */
     private String _clientId = null;
 
     /***
      * The session identifier uniquely distinguishes this session from any
      * other session. No two sessions can have the same session identifier.
      * This is allocated by the connection that created this session
      */
     private String _sessionId = null;
 
     /***
      * Back pointer to the connection that created this session. This
      * is set during object creation time
      */
     private JmsServerConnection _connection = null;
 
     /***
      * Maintain a list of consumers along with their associated destinations
      * that have been created through this session
      */
     private HashMap _consumers = new HashMap();
 
     /***
      * The message listener is the reference to a remote client that wull
      * receive the messages
      */
     private JmsMessageListener _listener = null;
 
     /***
      * This is the acknowledgement mode for the session
      */
     private int _ackMode = Session.AUTO_ACKNOWLEDGE;
 
     /***
      * Indicates whether the session is transactional
      */
     private boolean _transacted = false;
 
     /***
      * Holds the current xid that this session is associated with. A session
      * can olny be associated with one xid at any one time.
      */
     private Xid _xid = null;
 
     /***
      * Indicates if the underlying connection of this session has been stopped
      */
     private boolean _stopped = true;
 
     /***
      * Indicated that the session has been closed
      */
     private boolean _closed = false;
 
     /***
      * Holds the number of messages published by this session
      */
     private long _publishCount;
 
     /***
      * Holds the number of messages consumed by this session
      */
     private long _consumeCount;
 
     /***
      * Caches all sent messages
      */
     private SentMessageCache _sentMessageCache;
 
     /***
      * The logger
      */
     private static final Log _log = LogFactory.getLog(JmsServerSession.class);
 
 
     /***
      * Create an instance of the session using the specified client identifier.
      * also pass a back pointer to the connection that created this session.
      * By default, the session is stopped. The start() method must be invoked
      * before messages will be dispatched to consumers.
      *
      * @param       connection  connection that created this session
      * @param       id          client identity
      * @param       int         acknowledgement mode for the session
      * @param       transacted  true if the session is transactional
      */
     JmsServerSession(JmsServerConnection connection, String id, int ackmode,
                      boolean transacted) {
         _connection = connection;
         _clientId = id;
         _ackMode = ackmode;
         _transacted = transacted;
         _stopped = true;
         _sentMessageCache = new SentMessageCache(this);
     }
 
     /***
      * Set the session id for this object. The Connection is responsible for
      * setting the session id once it has been created
      *
      * @param       id          session id
      */
     void setSessionId(String id) {
         _sessionId = id;
     }
 
     /***
      * Return a reference to the client id
      *
      * @return      String      client id
      */
     public String getClientId() {
         return _clientId;
     }
 
     /***
      * Return a reference to the session id
      *
      * @return      String      session id
      */
     public String getSessionId() {
         return _sessionId;
     }
 
     /***
      * Start the message delivery for the session. If session delivery has
      * already started then treat the operation as an no-op
      */
     public void start() {
         if (_log.isDebugEnabled()) {
             _log.debug("start() [sessionId=" + _sessionId + "]");
         }
 
         if (_stopped) {
             pause(false);
             _stopped = false;
         }
     }
 
     /***
      * Stop message delivery for the session
      */
     public void stop() {
         if (_log.isDebugEnabled()) {
             _log.debug("stop() [sessionId=" + _sessionId + "]");
         }
         if (!_stopped) {
             pause(true);
             _stopped = true;
         }
     }
 
     /***
      * Close and release any resource allocated to this session.
      * This method may be called by multiple threads.
      *
      * @throws JMSException if the session cannot be closed
      */
     public void close() throws JMSException {
         boolean closed = false;
 
         synchronized (this) {
             closed = _closed;
             if (!closed) {
                 _closed = true;
             }
         }
 
         if (!closed) {
             if (_log.isDebugEnabled()) {
                 _log.debug("close() [sessionId=" + _sessionId + "]");
             }
 
             // reset the listener
             setMessageListener(null);
 
             // iterate over the list of consumers and deregister the
             // associated endpoints and then remove all the entries
             Iterator consumers = _consumers.values().iterator();
             while (consumers.hasNext()) {
                 ConsumerEndpoint consumer =
                     (ConsumerEndpoint) consumers.next();
                 ConsumerManager.instance().deleteConsumerEndpoint(consumer);
             }
 
             // clear the unacked message cache
             _sentMessageCache.clear();
 
             // clear the consumers
             _consumers.clear();
 
             // de-register the session from the connection
             _connection.deleteSession(this);
         } else {
             if (_log.isDebugEnabled()) {
                 _log.debug("close() [sessionId=" + _sessionId +
                     "]: session already closed");
             }
         }
     }
 
     /***
      * Acknowledge that the message with the following id has been processed
      *
      * @param clientId the clientId that sent the message to the client
      * @param id the message to ack
      * @throws JMSException if message cannot be acknowledged
      */
     public void acknowledgeMessage(long clientId, String id)
         throws JMSException {
         _sentMessageCache.acknowledgeMessage(new MessageId(id), clientId);
     }
 
     /***
      * Send the specified message to the server
      *
      * @param message the message to send
      * @throws JMSException if the message can't be sent
      */
     public void sendMessage(Message message) throws JMSException {
         if (message == null) {
             throw new JMSException("Message is null");
         }
 
         try {
             // check the delivery mode of the message
             checkDeliveryMode((MessageImpl) message);
 
             // set the connection identity and then let the message meanager
             // process it
             ((MessageImpl) message).setConnectionId(_connection.hashCode());
 
             // if there is a global transaction currently in process then
             // we must send the message to the resource manager, otherwise
             // send it directly to the message manager
             if (_xid != null) {
                 ResourceManager.instance().logPublishedMessage(_xid,
                     (MessageImpl) message);
             } else {
                 MessageMgr.instance().add((MessageImpl) message);
                 _publishCount++;
             }
         } catch (JMSException exception) {
             _log.error("Failed to process message", exception);
             throw exception;
         } catch (OutOfMemoryError exception) {
             String msg =
                 "Failed to process message due to out-of-memory error";
             _log.error(msg, exception);
             throw new JMSException(msg);
         } catch (Exception exception) {
             String msg = "Failed to process message";
             _log.error(msg, exception);
             throw new JMSException(msg);
         }
     }
 
     /***
      * Send the specified messages to the server.
      *
      * @param messages the messages to send
      * @throws JMSException if the messages can't be sent
      */
     public void sendMessages(Vector messages) throws JMSException {
         if (messages == null) {
             throw new JMSException("No messages to send");
         }
 
         MessageImpl message = null;
         while ((messages.size() > 0) &&
             ((message = (MessageImpl) messages.remove(0)) != null)) {
             try {
                 // check the delivery mode of the message
                 checkDeliveryMode((MessageImpl) message);
 
                 // set the connection identity and then let the message manager
                 // process it
                 message.setConnectionId(_connection.hashCode());
 
                 // if there is a global transaction in progress then send the
                 // message to the resource manager, otherwise send it to the
                 // message manager
                 if (_xid != null) {
                     ResourceManager.instance().logPublishedMessage(_xid,
                         message);
                 } else {
                     MessageMgr.instance().add(message);
                     _publishCount++;
                 }
             } catch (JMSException exception) {
                 _log.error("Failed to process message", exception);
                 throw exception;
             } catch (OutOfMemoryError exception) {
                 String msg =
                     "Failed to process message due to out-of-memory error";
                 _log.error(msg, exception);
                 throw new JMSException(msg);
             } catch (Exception exception) {
                 String msg = "Failed to process messages";
                 _log.error(msg, exception);
                 throw new JMSException(msg);
             }
         }
     }
 
     /***
      * Return the next message for the specified client. The <code>wait</code>
      * parameter indicates how long many milliseconds to wait for a message
      * before returning. If <code>wait</code> is 0 then do not wait at all. If
      * <code>wait</code> is -1 then wait indefinitely for the next message
      *
      * @param clientId the client identity
      * @param wait number of ms to wait
      * @return the next message or <code>null</code> if there is no message
      * @throws JMSException if the message can't be received
      */
     public Message receiveMessage(long clientId, long wait)
         throws JMSException {
         MessageImpl message = null;
         ConsumerEndpoint consumer = getConsumerEndpoint(clientId);
         if (consumer == null) {
             throw new JMSException(
                 "Can't receive message: no consumer registered with "
                 + "identifier " + clientId + " on session " + _sessionId);
         }
 
         // we have a valid consumer, now we need retrieve a handle.
         MessageHandle handle = consumer.receiveMessage(wait);
         if (handle != null) {
             // if we get a non-null handle the retrieve the message,
             // clone
             MessageImpl orig = handle.getMessage();
             if (orig != null) {
                 try {
                     message = (MessageImpl) orig.clone();
                     message.setJMSRedelivered(handle.getDelivered());
                     message.setClientId(handle.getClientId());
                     _consumeCount++;
                 } catch (Exception exception) {
                     _log.error(exception);
                     message = null;
                 }
             }
         }
 
         // if we have a non-null message then add it to the sent message
         // cache. Additionally, if we are part of a global transaction then
         // we must also sent it to the ResourceManager for recovery.
         if (handle != null) {
             _sentMessageCache.process(handle);
 
             if (_xid != null) {
                 try {
                     ResourceManager.instance().logReceivedMessage(
                         _xid, consumer.getId(), handle);
                 } catch (Exception exception) {
                     _log.error(exception);
                     JMSException jms_exception = new JMSException(
                         "Error in receiveMessage");
                     jms_exception.setLinkedException(exception);
                     throw jms_exception;
                 }
             }
         }
 
         return message;
     }
 
     /***
      * Return up to count messages from the endpoint with the specified
      * client identity. The client must be a QueueBrowser.
      *
      * @param       clientId            the client identity
      * @param       count               the maximum number of messages retrieve
      * @return      Message             the next message or null
      * @throws JMSException if the endpoint does not exist, or is not a
      * {@link QueueBrowserEndpoint}
      */
     public Vector receiveMessages(long clientId, int count)
         throws JMSException {
 
         ConsumerEndpoint consumer = getConsumerEndpoint(clientId);
         if (consumer == null) {
             throw new JMSException(
                 "Can't receive messages: no consumer registered with "
                 + "identifier " + clientId + " on session " + _sessionId);
         }
 
         if (!(consumer instanceof QueueBrowserEndpoint)) {
             throw new JMSException(
                 "Can't receive messages: consumer with identifier "
                 + "identifier " + clientId + " is not a QueueBrowser");
         }
 
         // we have a valid consumer, now we need retrieve upto count
         // handles
         Vector handles = ((QueueBrowserEndpoint) consumer).receiveMessages(
             count);
         Vector messages = new Vector();
         if (handles.size() > 0) {
 
             // process the handles
             int max = handles.size();
             for (int index = 0; index < max; index++) {
                 MessageHandle handle = (MessageHandle) handles.elementAt(index);
                 MessageImpl orig = handle.getMessage();
                 MessageImpl message = null;
                 if (orig != null) {
                     try {
                         message = (MessageImpl) orig.clone();
                         message.setJMSRedelivered(handle.getDelivered());
                         message.setClientId(handle.getClientId());
                         messages.addElement(message);
                     } catch (Exception exception) {
                         _log.error(exception);
                         message = null;
                     }
                 }
             }
         }
 
         return messages;
     }
 
     /***
      * Create an amdinistered queue, through the message manager admin
      * interface.
      *
      * @param queue administered queue to create
      * @throws JMSException if the queue can't be created
      */
     public void createQueue(JmsQueue queue) throws JMSException {
         if (!DestinationManager.instance().createAdministeredDestination(
             queue)) {
             throw new JMSException("Failed to create queue: " +
                 queue.getName());
         }
     }
 
     /***
      * Create an administered topic, through the message manager admin
      * interface.
      *
      * @param topic administered topic to create
      * @throws JMSException if the topic can't be created
      */
     public void createTopic(JmsTopic topic) throws JMSException {
         if (!DestinationManager.instance().createAdministeredDestination(
             topic)) {
             throw new JMSException("Failed to create topic: " +
                 topic.getName());
         }
     }
 
     /***
      * Create a receiver endpoint for this session. A receiver is a message
      * consumer specific to the queue message model. The receiver is
      * associated with a queue.
      * <p>
      * You cannot create more than one receiver for the same destination
      *
      * @param queue the receiver destination
      * @param consumerId the client session allocated identifier of the
      * consumer
      * @param selector the selector to filter messages. May be
      * <code>null</code>
      * @throws JMSException if the receiver can't be created
      */
     public void createReceiver(JmsQueue queue, long clientId, String selector)
         throws JMSException {
         if (_log.isDebugEnabled()) {
             _log.debug("createReceiver(queue=" + queue + ", clientId="
                 + clientId + ", selector=" + selector
                 + ") [sessionId=" + _sessionId + "]");
         }
 
         if (queue == null) {
             throw new JMSException("Cannot create receiver for null queue");
         }
 
         // Retrieve the destination from the destination manager and use
         // it to create the consumer
         ConsumerEndpoint consumer =
             ConsumerManager.instance().createConsumerEndpoint(this,
                 clientId, queue, selector);
         consumer.setAckMode(_ackMode);
         consumer.setConnectionId(_connection.hashCode());
         consumer.setTransacted(_transacted);
         // if the session is stopped then we should also stop the
         // consumer, so that it doesn't deliver messages and then
         // cache it for future reference.
         consumer.setStopped(_stopped);
         _consumers.put(Long.toString(clientId), consumer);
     }
 
     /***
      * This is a no-op
      */
     public void createSender(JmsQueue queue) throws JMSException {
     }
 
     /***
      * Create a queue browser for this session. This allows clients to browse
      * a queue without removing any messages.
      * <p>
      *
      * You cannot create more than one queue browser for the same queue
      * in a single session.
      *
      * @param       queue               queue to browse
      * @param       clientId            the client identity
      * @param       selector            message selector. This may be null
      * @throws   JMSException.
      */
     public void createBrowser(JmsQueue queue, long clientId, String selector)
         throws JMSException {
         if (_log.isDebugEnabled()) {
             _log.debug("createBrowser(queue=" + queue + ", clientId="
                 + clientId + ", selector=" + selector
                 + ") [sessionId=" + _sessionId + "]");
         }
 
         // check to see that we have a valid queue
         if (queue == null) {
             throw new JMSException("Cannot create browser for null queue");
         }
 
         // Retrieve the destination from the destination manager and use
         // it to create the consumer
         ConsumerEndpoint consumer =
             ConsumerManager.instance().createQueueBrowserEndpoint(this,
                 clientId, queue, selector);
 
         // if the session is stopped then we should also stop the
         // consumer, so that it doesn't deliver messages and then
         // cache it for future reference.
         consumer.setStopped(_stopped);
         _consumers.put(Long.toString(clientId), consumer);
     }
 
     /***
      * Delete the receiver with the specified identity and clean up all
      * associated resources.
      *
      * @param clientId the identity of the receiver
      * @throws JMSException if the receiver cannot be deleted
      */
     public void deleteReceiver(long clientId) throws JMSException {
         if (_log.isDebugEnabled()) {
             _log.debug("deleteReceiver(clientId=" + clientId + ") [sessionId="
                 + _sessionId + "]");
         }
 
         ConsumerEndpoint consumer =
             (ConsumerEndpoint) _consumers.remove(Long.toString(clientId));
         if (consumer == null) {
             throw new JMSException("No receiver with id " + clientId);
         }
 
         // destroy the consumer endpoint
         ConsumerManager.instance().deleteConsumerEndpoint(consumer);
     }
 
     /***
      * Delete the sender associated with the specified queue from the session
      * If the corresponding sender does not exist or it cannot delete it then
      * throw the JMSException.
      *
      * @param clientId the identity of the sender
      * @throws JMSException if the sender cannot be deleted
      */
     public void deleteSender(long clientId) throws JMSException {
         // no-op
     }
 
     /***
      * Delete the queue browser associated with the specified queue from
      * the session.
      *
      * @param clientId the identity of the browser
      * @throws JMSException if the browser cannot be deleted
      */
     public void deleteBrowser(long clientId) throws JMSException {
         ConsumerEndpoint consumer =
             (ConsumerEndpoint) _consumers.remove(Long.toString(clientId));
         if (consumer == null) {
             throw new JMSException("No browser with id " + clientId);
         }
         // destroy the consumer endpoint
         ConsumerManager.instance().deleteConsumerEndpoint(consumer);
     }
 
     /***
      * Create a subscriber endpoint for this session. A subscriber is a message
      * consumer specific to the topic message model. The subscriber is
      * associated with a topic. Register the consumer with the message
      * manager so that a queue can be set up for it. Finally add the consumer
      * to the list of consumers managed by this session.
      * <p>
      * Note that the message manager manages consumers  for all server sessions
      * <p>
      * You cannot create more than one subscriber for the same destination.
      * Currently we don't check this
      *
      * @param       topic               subscriber destination
      * @param       name                consumer name
      * @param       clientId            the client session allocated
      *                                  identifier of the consumer
      * @param       selector            the selector to filter messages.
      *                                  This may be null.
      * @param       noLocal             true to inhibit consumption of messages
      *                                  published on this connection.
      * @throws   JMSException.
      */
     public void createSubscriber(JmsTopic topic, String name, long clientId,
                                  String selector, boolean noLocal)
         throws JMSException {
 
         if (_log.isDebugEnabled()) {
             _log.debug("createSubscriber(topic=" + topic + ", name=" + name
                 + ", clientId=" + clientId + ", selector=" + selector
                 + ", noLocal=" + noLocal + ") [sessionId="
                 + _sessionId + "]");
         }
 
         // check to ensure that the methods preconditions have been met
         if (topic == null) {
             throw new JMSException("Cannot create subscriber for null topic");
         }
 
         // Retrieve the destination from the destination manager and
         // use it to create the consumer through the consumer manager.
         ConsumerEndpoint consumer = null;
 
         if (name != null) {
             if (name.length() > 0) {
 
                 // for a durable consumer the topic must be
                 ConsumerManager manager = ConsumerManager.instance();
 
                 if (manager.durableConsumerExists(name)) {
                     // if the durable consumer exists then validate that
                     // it was the specified topic that it was registered
                     // under. If it is not registered for the topic then
                     // we must delete the existing entry and recreate it
                     // against the new topic
                     if (!manager.validSubscription(topic.getName(), name)) {
                         unsubscribe(name);
                         manager.createDurableConsumer(topic, name);
                     }
                 } else {
                     // if the durable consumer does not exist then create
                     // it
                     manager.createDurableConsumer(topic, name);
                 }
 
                 // if a durable subscriber with the specified name is
                 // alreayd active then this method will throw an
                 // exception.
                 // attempt to create a durable consuinmer
                 consumer = manager.createDurableConsumerEndpoint(this,
                     topic, name, clientId, selector);
                 consumer.setConnectionId(_connection.hashCode());
                 consumer.setTransacted(_transacted);
                 consumer.setAckMode(_ackMode);
                 consumer.setNoLocal(noLocal);
             } else {
                 throw new JMSException("Name in createSubscriber was null");
             }
         } else {
             // Create a non-durable subscriber for the specified destination
             // and using the required selector.
             consumer = ConsumerManager.instance().createConsumerEndpoint(this,
                 clientId, topic, selector);
             consumer.setConnectionId(_connection.hashCode());
             consumer.setTransacted(_transacted);
             consumer.setAckMode(_ackMode);
             consumer.setNoLocal(noLocal);
         }
 
         // once the consumer has been created then set it to the same state
         // as the session and add it to the list on consumers to manage
         consumer.setStopped(_stopped);
         _consumers.put(Long.toString(clientId), consumer);
     }
 
     /***
      * This should be a no operation. Do we need to maintain state information
      * that a publisher has been created.
      *
      * @param       topic               receiver destination
      * @throws   JMSException.
      */
     public void createPublisher(JmsTopic topic)
         throws JMSException {
     }
 
     /***
      * This function deletes a persistent subsrciber and its history from
      * the database. It his subscriber re-connects it get everything available
      * for the queue topic. If the subscriber is reliable, this is a no op.
      * See UnregisterSubscriber below for just unregistering the subscriber
      * but leaving its persistent data in the db.
      * <p>
      * The data contains information necessary to delete the subscriber
      *
      * @param       clientId            the client identity
      * @throws   JMSException.
      */
     public void deleteSubscriber(long clientId) throws JMSException {
         if (_log.isDebugEnabled()) {
             _log.debug("deleteSubscriber(clientId=" + clientId
                 + ") [sessionId=" + _sessionId + "]");
         }
 
         // retrieve the endpoint corresponding to the client id and
         // then acknowledge the messsage
         ConsumerEndpoint consumer =
             (ConsumerEndpoint) _consumers.remove(Long.toString(clientId));
         if (consumer == null) {
             throw new JMSException("Failed to close consumer with id " +
                 "[" + hashCode() + ":" + clientId + "]");
         }
 
         ConsumerManager.instance().deleteConsumerEndpoint(consumer);
     }
 
     /***
      * Delete the publisher associated with the specified topic from the
      * session. If the corresponding publisher does not exist or it cannot
      * delete it then throw the JMSException.
      *
      * @param       topic               sender destination
      * @throws   JMSException.
      */
     public void deletePublisher(JmsTopic topic) throws JMSException {
         // no-op
     }
 
     /***
      * Unsubscribe a durable subscription. This will delete the state of
      * the durable subscriber maintained by the server. A durable subscriber
      * is uniquely identifiable and the same subscriber cannot be associated
      * with more than topic.
      *
      * @param       name                the name used to uniquely identify the
      *                                  subscription
      * @throws   JMSException        if the subscription cannot be removed
      *                                  or any other problem.
      */
     public void unsubscribe(String name) throws JMSException {
         if (_log.isDebugEnabled()) {
             _log.debug("unsubscribe(name=" + name + ") [sessionId="
                 + _sessionId + "]");
         }
 
         ConsumerManager manager = ConsumerManager.instance();
 
         // check that the durable consumer actually exists. If it doesn't then
         // throw an exception
         if (!manager.durableConsumerExists(name)) {
             throw new InvalidDestinationException(name +
                 " is not a durable subscriber name");
         }
 
         // check that the durable consumer is not active before removing it. If
         // it is then throw an exception
         if (!manager.isDurableConsumerActive(name)) {
             manager.removeDurableConsumer(name);
         } else {
             throw new JMSException("Failed to unsubscribe subscriber "
                 + name + " since is still active");
         }
     }
 
     /***
      * Stop message delivery to this session. If there are any problems
      * completing the request then throw the JMSException exception
      *
      * @throws   JMSException
      */
     public void stopMessageDelivery() throws JMSException {
         stop();
     }
 
     /***
      * Start message delivery to this session. If there are any problems
      * completing this request then throw the JMSException exception
      *
      * @throws   JMSException
      */
     public void startMessageDelivery() throws JMSException {
         start();
     }
 
     /***
      * Check if the specified message handle is in the session's list
      * of unacked messages
      *
      * @param handle - the handle to query
      * @return boolean - true if it is and false otherwise
      */
     public boolean containsUnackedHandle(MessageHandle handle) {
         return _sentMessageCache.handleInCache(handle);
     }
 
     // implementation of InternalMessageListener.onMessage
     public void onMessage(MessageHandle handle, boolean ignore)
         throws Exception {
 
         if ((handle != null) &&
             (_listener != null)) {
             MessageImpl message = handle.getMessage();
             MessageImpl m = null;
             if (message != null) {
                 m = (MessageImpl) message.clone();
                 m.setClientId(handle.getClientId());
                 m.setJMSRedelivered(handle.getDelivered());
 
                 // if we are acking the message and the session is
                 // transacted and the acknowledge mode is
                 // CLIENT_ACKNOWLEDGE then send it to the cache before
                 // we send it to the listener. This will enable clients
                 // to ack the message while in the onMessage method
                 if ((_transacted) ||
                     (_ackMode == Session.CLIENT_ACKNOWLEDGE)) {
                     _sentMessageCache.process(handle);
                 }
 
                 try {
                     // send the message to the listener.
                     _listener.onMessage(m);
 
                     // if the session is not transacted or the acknowledge mode
                     // is not CLIENT_ACKNOWLEDGE then process it through the
                     // sent message cache now.
                     if ((!_transacted) &&
                         (_ackMode != Session.CLIENT_ACKNOWLEDGE)) {
                         _sentMessageCache.process(handle);
                     }
                 } catch (ClientDisconnectionException exception) {
                     // close all resources and rethrow it
                     close();
                     throw exception;
                 }
             } else {
                 throw new JMSException(
                     "Could not get message for handle " + handle,
                     JMSErrorCodes.FailedToResolveHandle);
             }
         }
     }
 
     // implementation of InternalMessageListener.onMessage
     public void onMessages(Vector handles) throws Exception {
         _log.error("Illegal to call onMessage");
         Thread.currentThread().dumpStack();
     }
 
     // implementation of InternalMessageListener.onMessageAvailable
     public void onMessageAvailable(long clientId) throws Exception {
         _listener.onMessageAvailable(clientId);
     }
 
     /***
      * This will send a null message down the connection to the client to
      * test whether the client endpoint is alive.
      *
      * @return <code>true</code> if it is active, otherwise <code>false</code>
      */
     public boolean isClientEndpointActive() {
         boolean active = true;
         if (_listener != null) {
             try {
                 // send the message to the listener.
                 _listener.onMessage(null);
             } catch (ClientDisconnectionException exception) {
                 _log.info("Failed to verify that session " + _sessionId
                     + " is active.");
                 active = false;
                 // ignore the exception
             }
         }
 
         return active;
     }
 
     /***
      * Set a message listener for the session. This is the channel used
      * to asynchronously deliver messages to consumers created on this
      * session.
      *
      * @param listener the message listener
      */
     public void setMessageListener(JmsMessageListener listener) {
         _listener = listener;
     }
 
     /***
      * Check whether to enable asynchronous message delivery for a particular
      * consumer
      *
      * @param clientId the id of the client to check
      * @param id the last processed message
      * @param enable <code>true</code> to enable; <code>false</code> to disable
      */
     public void enableAsynchronousDelivery(long clientId, String id,
                                            boolean enable)
         throws JMSException {
         ConsumerEndpoint consumer = getConsumerEndpoint(clientId);
         if (consumer == null) {
             throw new JMSException(clientId + " is not registered");
         }
 
         if (enable) {
             consumer.setMessageListener(this);
         } else {
             consumer.setMessageListener(null);
         }
     }
 
     /***
      * Call recover on all registered consumers. This will cause all
      * unacknowledged messages to be redelivered. Before we recover we
      * need to stop messages delivery. We then need to start redelivery
      * when the recovery has been completed
      *
      * @throws JMSException if the session can't be recovered
      */
     public void recover() throws JMSException {
         // stop message delivery
         stop();
 
         // iterate over the list of consumers recover them
         Iterator consumers = _consumers.values().iterator();
         while (consumers.hasNext()) {
             ((ConsumerEndpoint) consumers.next()).recover();
         }
 
         // clear the messages in the sent message cache
         _sentMessageCache.clear();
 
         // restart message delivery
         start();
 
 
     }
 
     /***
      * Commit this session, which will acknowledge all sent messages for
      * all consumers.
      *
      * @throws JMSException - if there are any problems
      */
     public void commit() throws JMSException {
         try {
             _sentMessageCache.acknowledgeAllMessages();
         } catch (OutOfMemoryError exception) {
             String msg =
                 "Failed to commit transaction due to out-of-memory error";
             _log.error(msg, exception);
             throw new JMSException(msg);
         }
     }
 
     /***
      * Abort, will return all unacked messages to their respective endpoints,
      * if they are still active.
      *
      * @throws JMSException - if there are any problems
      */
     public void rollback()
         throws JMSException {
         _sentMessageCache.clear();
     }
 
     // implementation XAResource.commit
     public void commit(Xid xid, boolean onePhase) throws XAException {
         try {
             ResourceManager.instance().commit(xid, onePhase);
         } catch (ResourceManagerException exception) {
             throw new XAException("Failed in commit " + exception);
         } finally {
             _xid = null;
         }
     }
 
     // implementation of XAResource.end
     public void end(Xid xid, int flags) throws XAException {
         try {
             ResourceManager.instance().end(xid, flags);
         } catch (ResourceManagerException exception) {
             throw new XAException("Failed in end " + exception);
         } finally {
             _xid = null;
         }
     }
 
     // implementation of XAResource.forget
     public void forget(Xid xid) throws XAException {
         try {
             ResourceManager.instance().forget(xid);
         } catch (ResourceManagerException exception) {
             throw new XAException("Failed in forget " + exception);
         } finally {
             _xid = null;
         }
     }
 
     // implementation of XAResource.getTransactionTimeout
     public int getTransactionTimeout() throws XAException {
         try {
             return ResourceManager.instance().getTransactionTimeout();
         } catch (ResourceManagerException exception) {
             throw new XAException("Failed in getTransactionTimeout " +
                 exception);
         }
     }
 
     // implementation of XAResource.isSameRM
     public boolean isSameRM(XAResource xares) throws XAException {
         return true;
     }
 
     // implementation XAResource.isSame
     public int prepare(Xid xid) throws XAException {
         try {
             return ResourceManager.instance().prepare(xid);
         } catch (ResourceManagerException exception) {
             throw new XAException("Failed in prepare " + exception);
         }
     }
 
     // implementation of XAResource.prepare
     public Xid[] recover(int flag) throws XAException {
         try {
             return ResourceManager.instance().recover(flag);
         } catch (ResourceManagerException exception) {
             throw new XAException("Failed in recover " + exception);
         }
     }
 
     // implementation of XAResource.recover
     public void rollback(Xid xid) throws XAException {
         try {
             ResourceManager.instance().rollback(xid);
         } catch (ResourceManagerException exception) {
             throw new XAException("Failed in rollback " + exception);
         } finally {
             // clear the current xid
             _xid = null;
         }
     }
 
     // implementation of XAResource.rollback
     public boolean setTransactionTimeout(int seconds) throws XAException {
         try {
             return ResourceManager.instance().setTransactionTimeout(seconds);
         } catch (ResourceManagerException exception) {
             throw new XAException("Failed in setTransactionTimeout "
                 + exception);
         }
     }
 
     // implementation of XAResource.setTransactionTimeout
     public void start(Xid xid, int flags) throws XAException {
         try {
             ResourceManager.instance().start(xid, flags);
 
             // set this as the current xid for this session
             _xid = xid;
         } catch (ResourceManagerException exception) {
             throw new XAException("Failed in start " + exception);
         }
     }
 
     /***
      * Return the xid that is currently associated with this session or null
      * if this session is currently not part of a global transactions
      *
      * @return Xid
      */
     public Xid getXid() {
         return _xid;
     }
 
     /***
      * Return the identity of the {@link ResourceManager}. The transaction
      * manager should be the only one to initiating this call.
      *
      * @return the identity of the resource manager
      * @throws XAException - if it cannot retrieve the rid.
      */
     public String getResourceManagerId() throws XAException {
         try {
             return ResourceManager.instance().getResourceManagerId();
         } catch (ResourceManagerException exception) {
             throw new XAException("Failed in getResourceManagerId "
                 + exception);
         }
     }
 
     /***
      * Determines if the session is transacted
      *
      * @return <code>true</code> if the session is transacted
      */
     public boolean isTransacted() {
         return _transacted;
     }
 
     /***
      * Returns the message acknowledgement mode for the session
      */
     public int getAckMode() {
         return _ackMode;
     }
 
     /***
      * Returns the consumer endpoint for the supplied client id
      *
      * @param clientId the identity of the consumer endpoint
      * @return the consumer endpoint corresponding to <code>clientId</code>,
      * or <code>null</code> if none exists
      */
     public ConsumerEndpoint getConsumerEndpoint(long clientId) {
         String identity = Long.toString(clientId);
         return (ConsumerEndpoint) _consumers.get(identity);
     }
 
     /***
      * This method is used to stop and restart the session. Stopping the
      * session should stop all message delivery to session consumers
      *
      * @param stop - true if we need to stop the session, false otherwise
      */
     private void pause(boolean stop) {
         Iterator iter = _consumers.values().iterator();
         while (iter.hasNext()) {
             ((ConsumerEndpoint) iter.next()).setStopped(stop);
         }
     }
 
     /***
      * Check the delivery mode of the message. If the delivery mode is
      * persistent and the destination is non-administered then change the
      * delivery mode to non-persistent so that it can be processed correctly
      * by the server
      *
      * @param message - the message to check
      * @throws JMSException - propagate JMSException to client
      */
     private void checkDeliveryMode(MessageImpl message) throws JMSException {
         if ((message.getJMSDeliveryMode() == DeliveryMode.PERSISTENT) &&
             (!DestinationManager.instance().isMessageForAdministeredDestination(message))) {
             message.setJMSDeliveryMode(DeliveryMode.NON_PERSISTENT);
         }
     }
 
 } //-- JmsServerSession