pooledspringxasession.java

Jencks是一个轻量级的JCA容器。它可以轻松部署到Spring中以提供消息驱动的POJOs.此外Jencks通过使用类似于JMS,JAX-RPC,JBI与JCA CCI的API来提供对inboun

/** 
 * Licensed under the Apache License, Version 2.0 (the "License"); 
 * you may not use this file except in compliance with the License. 
 * You may obtain a copy of the License at 
 * 
 * http://www.apache.org/licenses/LICENSE-2.0
 * 
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS, 
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 
 * See the License for the specific language governing permissions and 
 * limitations under the License. 
 * 
 **/
package org.jencks.pool;

import java.io.Serializable;

import javax.jms.XASession;
import javax.jms.TopicSession;
import javax.jms.QueueSession;
import javax.jms.JMSException;
import javax.jms.BytesMessage;
import javax.jms.MapMessage;
import javax.jms.Message;
import javax.jms.ObjectMessage;
import javax.jms.Queue;
import javax.jms.StreamMessage;
import javax.jms.TemporaryQueue;
import javax.jms.TemporaryTopic;
import javax.jms.TextMessage;
import javax.jms.Topic;
import javax.jms.QueueBrowser;
import javax.jms.MessageConsumer;
import javax.jms.Destination;
import javax.jms.TopicSubscriber;
import javax.jms.MessageListener;
import javax.jms.QueueReceiver;
import javax.jms.MessageProducer;
import javax.jms.QueueSender;
import javax.jms.TopicPublisher;
import javax.jms.Session;
import javax.transaction.xa.XAResource;

import org.apache.commons.pool.ObjectPool;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

public class PooledSpringXASession implements TopicSession, QueueSession, XASession
{
  private static final Log log = LogFactory.getLog(PooledSpringXASession.class);
  
  private XASession session;
  private ObjectPool sessionPool;
  private MessageProducer messageProducer;
  private QueueSender queueSender;
  private TopicPublisher topicPublisher;
  private boolean ignoreClose;

  public PooledSpringXASession(final XASession session, final ObjectPool sessionPool)
  {
    this.session = session;
    this.sessionPool = sessionPool;
    this.ignoreClose = false;
  }

  public boolean getIgnoreClose()
  {
    return this.ignoreClose;
  }

  public void setIgnoreClose(final boolean ignoreClose)
  {
    this.ignoreClose = ignoreClose;
  }

  /**
   * If the Session goes into an unstable (unusable) state, then we want to
   * close it down and permanently remove it from the pool.
   */
  public void destroyAndRemoveFromPool()
  {
    try {
      sessionPool.invalidateObject(this);
    } catch (Throwable t) {
      log.warn("Unable to remove invalidated JMS Session from the pool due to the following exception.  Will ignore the exception and continue.", t);
    }
  }


  public void close() throws JMSException
  {
    if(log.isDebugEnabled()) log.debug("---->>>>> PooledSpringXASession.close() called");
    // If we are associated with a transaction, then we will let
    // PooledSpringXAConnection's transaction synchronization handle closing
    // us at the end of the transaction.
    if(!getIgnoreClose()) {
      if(log.isDebugEnabled()) log.debug("---->>>>> ignoreClose = false, so returning session pool...");
      // TODO a cleaner way to reset??

      // lets reset the session
      getActualSession().setMessageListener(null);

      try {
        sessionPool.returnObject(this);
      }
      catch (Exception e) {
        final JMSException jmsException = new JMSException("Failed to return session to pool: " + e);
        jmsException.initCause(e);
        throw jmsException;
      }
    } else if(log.isDebugEnabled()) {
      log.debug("---->>>>> ignoreClose IS TRUE!  KEEPING SESSION OPEN!");
    }
  }

  public void commit() throws JMSException
  {
    throw new JMSException("Cannot commit() inside XASession");
  }

  public BytesMessage createBytesMessage() throws JMSException
  {
    return getActualSession().createBytesMessage();
  }

  public MapMessage createMapMessage() throws JMSException
  {
    return getActualSession().createMapMessage();
  }

  public Message createMessage() throws JMSException
  {
    return getActualSession().createMessage();
  }

  public ObjectMessage createObjectMessage() throws JMSException
  {
    return getActualSession().createObjectMessage();
  }

  public ObjectMessage createObjectMessage(Serializable serializable)
      throws JMSException
  {
    return getActualSession().createObjectMessage(serializable);
  }

  public Queue createQueue(String s) throws JMSException
  {
    return getActualSession().createQueue(s);
  }

  public StreamMessage createStreamMessage() throws JMSException
  {
    return getActualSession().createStreamMessage();
  }

  public TemporaryQueue createTemporaryQueue() throws JMSException
  {
    return getActualSession().createTemporaryQueue();
  }

  public TemporaryTopic createTemporaryTopic() throws JMSException
  {
    return getActualSession().createTemporaryTopic();
  }

  public void unsubscribe(String s) throws JMSException
  {
    getActualSession().unsubscribe(s);
  }

  public TextMessage createTextMessage() throws JMSException
  {
    return getActualSession().createTextMessage();
  }

  public TextMessage createTextMessage(String s) throws JMSException
  {
    return getActualSession().createTextMessage(s);
  }

  public Topic createTopic(String s) throws JMSException
  {
    return getActualSession().createTopic(s);
  }

  public int getAcknowledgeMode() throws JMSException
  {
    return getActualSession().getAcknowledgeMode();
  }

  public boolean getTransacted() throws JMSException
  {
    return true;
  }

  public void recover() throws JMSException
  {
    getActualSession().recover();
  }

  public void rollback() throws JMSException
  {
    throw new JMSException("Cannot rollback() inside XASession");
  }

  public void run()
  {
    if (session != null) {
      session.run();
    }
  }

  public XAResource getXAResource()
  {
    try {
      return getActualSession().getXAResource();
    } catch(Exception e) {
      throw new RuntimeException(e);
    }
  }

  public Session getSession() throws JMSException
  {
    return this;
  }


  // Consumer related methods
  //-------------------------------------------------------------------------
  public QueueBrowser createBrowser(Queue queue) throws JMSException
  {
    return getActualSession().createBrowser(queue);
  }

  public QueueBrowser createBrowser(Queue queue, String selector)
      throws JMSException
  {
    return getActualSession().createBrowser(queue, selector);
  }

  public MessageConsumer createConsumer(Destination destination)
      throws JMSException
  {
    return getActualSession().createConsumer(destination);
  }

  public MessageConsumer createConsumer(Destination destination,
                                        String selector) throws JMSException
  {
    return getActualSession().createConsumer(destination, selector);
  }

  public MessageConsumer createConsumer(Destination destination,
                                        String selector, boolean noLocal)
      throws JMSException
  {
    return getActualSession().createConsumer(destination, selector, noLocal);
  }

  public TopicSubscriber createDurableSubscriber(Topic topic, String selector)
      throws JMSException
  {
    return getActualSession().createDurableSubscriber(topic, selector);
  }

  public TopicSubscriber createDurableSubscriber(Topic topic, String name,
                                                 String selector,
                                                 boolean noLocal)
      throws JMSException
  {
    return getActualSession().createDurableSubscriber(topic, name, selector, noLocal);
  }

  public MessageListener getMessageListener() throws JMSException
  {
    return getActualSession().getMessageListener();
  }

  public void setMessageListener(MessageListener messageListener)
      throws JMSException
  {
    getActualSession().setMessageListener(messageListener);
  }

  public TopicSubscriber createSubscriber(Topic topic) throws JMSException
  {
    return ((TopicSession)getActualSession()).createSubscriber(topic);
  }

  public TopicSubscriber createSubscriber(Topic topic, String selector,
                                          boolean local) throws JMSException
  {
    return ((TopicSession)getActualSession()).createSubscriber(topic, selector, local);
  }

  public QueueReceiver createReceiver(Queue queue) throws JMSException
  {
    return ((QueueSession)getActualSession()).createReceiver(queue);
  }

  public QueueReceiver createReceiver(Queue queue, String selector)
      throws JMSException
  {
    return ((QueueSession)getActualSession()).createReceiver(queue, selector);
  }


  // Producer related methods
  //-------------------------------------------------------------------------
  public MessageProducer createProducer(Destination destination)
      throws JMSException
  {
    return new PooledProducer(getMessageProducer(), destination);
  }

  public QueueSender createSender(Queue queue) throws JMSException
  {
    return new PooledQueueSender(getQueueSender(), queue);
  }

  public TopicPublisher createPublisher(Topic topic) throws JMSException
  {
    return new PooledTopicPublisher(getTopicPublisher(), topic);
  }

  // Implementation methods
  //-------------------------------------------------------------------------
  public XASession getActualSession() throws JMSException
  {
    if (this.session == null) {
      throw new JMSException("The session has already been closed");
    }
    return this.session;
  }

  public MessageProducer getMessageProducer() throws JMSException
  {
    if (this.messageProducer == null) {
      this.messageProducer = getActualSession().createProducer(null);
    }
    return this.messageProducer;
  }

  public QueueSender getQueueSender() throws JMSException
  {
    if (this.queueSender == null) {
      this.queueSender = ((QueueSession)getActualSession()).createSender(null);
    }
    return this.queueSender;
  }

  public TopicPublisher getTopicPublisher() throws JMSException
  {
    if (this.topicPublisher == null) {
      this.topicPublisher = ((TopicSession)getActualSession()).createPublisher(null);
    }
    return this.topicPublisher;
  }

}