/*

  * Copyright (C) The Apache Software Foundation. All rights reserved.

  *

  * This software is published under the terms of the Apache Software License

  * version 1.1, a copy of which has been included with this distribution in

  * the LICENSE file.

  *

  * $Id: MessengerSupport.java 155459 2005-02-26 13:24:44Z dirkv $

  */

 package org.apache.commons.messenger;

 import java.io.Serializable;

 import java.util.HashMap;

 import java.util.Map;

 import javax.jms.BytesMessage;

 import javax.jms.Connection;

 import javax.jms.ConnectionConsumer;

 import javax.jms.ConnectionFactory;

 import javax.jms.DeliveryMode;

 import javax.jms.Destination;

 import javax.jms.JMSException;

 import javax.jms.MapMessage;

 import javax.jms.Message;

 import javax.jms.MessageConsumer;

 import javax.jms.MessageListener;

 import javax.jms.MessageProducer;

 import javax.jms.ObjectMessage;

 import javax.jms.Queue;

 import javax.jms.QueueBrowser;

 import javax.jms.QueueConnection;

 import javax.jms.QueueSender;

 import javax.jms.QueueSession;

 import javax.jms.ServerSessionPool;

 import javax.jms.Session;

 import javax.jms.StreamMessage;

 import javax.jms.TextMessage;

 import javax.jms.Topic;

 import javax.jms.TopicConnection;

 import javax.jms.TopicPublisher;

 import javax.jms.TopicSession;

 import org.apache.commons.logging.Log;

 import org.apache.commons.logging.LogFactory;

 /** <p><code>MessengerSupport</code> is an abstract base class which implements

   * most of the functionality of Messenger. Derivations need to specify the

   * connection and session creation and the pooling strategy.</p>

   *

   * @author <a href="mailto:jstrachan@apache.org">James Strachan</a>

   * @version $Revision: 155459 $

   */

 public abstract class MessengerSupport implements Messenger {

     /** Logger */

     private static final Log log = LogFactory.getLog(MessengerSupport.class);

     private static final Log destinationLog = LogFactory.getLog("org.apache.commons.messenger.destination");

     private static final boolean CACHE_REQUESTOR = true;

     /** The name of the Messenger */

     private String name;

     /** 

      * Whether Queue's and Topic's are looked up using JNDI (true)

      * or wether they should be created on the fly 

      */

     private boolean jndiDestinations;

     /** are topic subscribers durable? */

     private boolean durable;

     /** the delivery mode used by default when sending messages */

     private int deliveryMode = Message.DEFAULT_DELIVERY_MODE;

     /** the durable name used for durable topic based subscriptions */

     private String durableName;

     /** 

      * whether local messages are ignored when topic based subscription is used

      * with a message selector 

      */

     private boolean noLocal;

     /** Should we cache the requestor object per thread? */

     private boolean cacheRequestors;

     /** A Map of ListenerKey objects to MessageConsumer objects */

     private Map listeners = new HashMap();

     /** whether MessageProducer instances should be cached or not */

     private boolean cacheProducers = true;

     public MessengerSupport() {

     }

     public String toString() {

         try {

             MessengerSession session = borrowMessengerSession();

             String answer = super.toString() + " session: " + session.toString();

             returnMessengerSession(session);

             return answer;

         }

         catch (Exception e) {

             return super.toString() + " session: " + e.toString();

         }

     }

     public Destination getDestination(String subject) throws JMSException {

         MessengerSession messengerSession = borrowMessengerSession();

         try {

             boolean debug = destinationLog.isInfoEnabled();

             Session session = messengerSession.getSession();

             if (messengerSession.isTopic()) {

                 if (debug) {

                     destinationLog.info("Using topic: " + subject);

                 }

                 return getTopic((TopicSession) session, subject);

             }

             else {

                 if (debug) {

                     destinationLog.info("Using queue: " + subject);

                 }

                 return getQueue((QueueSession) session, subject);

             }

         }

         finally {

             returnMessengerSession(messengerSession);

         }

     }

     public Destination createTemporaryDestination() throws JMSException {

         MessengerSession messengerSession = borrowMessengerSession();

         try {

             Session session = messengerSession.getSession();

             if (messengerSession.isTopic()) {

                 TopicSession topicSession = (TopicSession) session;

                 return topicSession.createTemporaryTopic();

             }

             else {

                 QueueSession queueSession = (QueueSession) session;

                 return queueSession.createTemporaryQueue();

             }

         }

         finally {

             returnMessengerSession(messengerSession);

         }

     }

     public void send(Destination destination, Message message) throws JMSException {

         MessengerSession session = borrowMessengerSession();

         try {

             MessageProducer producer = session.getMessageProducer(destination);

             if (session.isTopic()) {

                 ((TopicPublisher) producer).publish((Topic) destination, message);

             }

             else {

                 ((QueueSender) producer).send((Queue) destination, message);

             }

         }

         finally {

             returnMessengerSession(session);

         }

     }

     public Message receive(Destination destination) throws JMSException {

         MessengerSession session = borrowMessengerSession();

         MessageConsumer consumer = null;

         try {

             consumer = borrowMessageConsumer(session, session.getSession(), destination);

             return consumer.receive();

         }

         finally {

             returnMessageConsumer(consumer);

             returnMessengerSession(session);

         }

     }

     public Message receive(Destination destination, String selector) throws JMSException {

         MessengerSession session = borrowMessengerSession();

         MessageConsumer consumer = null;

         try {

             consumer = borrowMessageConsumer(session, session.getSession(), destination, selector);

             return consumer.receive();

         }

         finally {

             returnMessageConsumer(consumer);

             returnMessengerSession(session);

         }

     }

     public Message receive(Destination destination, long timeoutMillis) throws JMSException {

         MessengerSession session = borrowMessengerSession();

         MessageConsumer consumer = null;

         try {

             consumer = borrowMessageConsumer(session, session.getSession(), destination);

             return consumer.receive(timeoutMillis);

         }

         finally {

             returnMessageConsumer(consumer);

             returnMessengerSession(session);

         }

     }

     public Message receive(Destination destination, String selector, long timeoutMillis) throws JMSException {

         MessengerSession session = borrowMessengerSession();

         MessageConsumer consumer = null;

         try {

             consumer = borrowMessageConsumer(session, session.getSession(), destination, selector);

             return consumer.receive(timeoutMillis);

         }

         finally {

             returnMessageConsumer(consumer);

             returnMessengerSession(session);

         }

     }

     public Message receiveNoWait(Destination destination) throws JMSException {

         MessengerSession session = borrowMessengerSession();

         MessageConsumer consumer = null;

         try {

             consumer = borrowMessageConsumer(session, session.getSession(), destination);

             return consumer.receiveNoWait();

         }

         finally {

             returnMessageConsumer(consumer);

             returnMessengerSession(session);

         }

     }

     public Message receiveNoWait(Destination destination, String selector) throws JMSException {

         MessengerSession session = borrowMessengerSession();

         MessageConsumer consumer = null;

         try {

             consumer = borrowMessageConsumer(session, session.getSession(), destination, selector);

             return consumer.receiveNoWait();

         }

         finally {

             returnMessageConsumer(consumer);

             returnMessengerSession(session);

         }

     }

     public MessageConsumer createConsumer(Destination destination) throws JMSException {

         MessengerSession session = borrowMessengerSession();

         try {

             return createMessageConsumer(session, session.getSession(), destination);

         }

         finally {

             returnMessengerSession(session);

         }

     }

     public MessageConsumer createConsumer(Destination destination, String selector) throws JMSException {

         MessengerSession session = borrowMessengerSession();

         try {

             return createMessageConsumer(session, session.getSession(), destination, selector);

         }

         finally {

             returnMessengerSession(session);

         }

     }

     public void run() {

         // don't return sessions which throw an exception

         try {

             MessengerSession session = borrowMessengerSession();

             session.getSession().run();

             returnMessengerSession(session);

         }

         catch (JMSException e) {

             // ### ignore

         }

     }

     public ConnectionConsumer createConnectionConsumer(

         Destination destination,

         ServerSessionPool sessionPool,

         int maxMessages)

         throws JMSException {

         return createConnectionConsumer(destination, null, sessionPool, maxMessages);

     }

     public ConnectionConsumer createConnectionConsumer(

         Destination destination,

         String selector,

         ServerSessionPool sessionPool,

         int maxMessages)

         throws JMSException {

         Connection connection = getConnection();

         if (isTopic(connection)) {

             TopicConnection topicConnection = (TopicConnection) connection;

             if (isDurable()) {

                 return topicConnection.createDurableConnectionConsumer(

                     (Topic) destination,

                     getDurableName(),

                     selector,

                     sessionPool,

                     maxMessages);

             }

             else {

                 return topicConnection.createConnectionConsumer(

                     (Topic) destination,

                     selector,

                     sessionPool,

                     maxMessages);

             }

         }

         else {

             QueueConnection queueConnection = (QueueConnection) connection;

             return queueConnection.createConnectionConsumer((Queue) destination, selector, sessionPool, maxMessages);

         }

     }

     public abstract Connection getConnection() throws JMSException;

     // Listener API

     //-------------------------------------------------------------------------

     public void addListener(Destination destination, MessageListener listener) throws JMSException {

         if (listener instanceof MessengerListener) {

             MessengerListener messengerListener = (MessengerListener) listener;

             messengerListener.setMessenger(this);

         }

         MessengerSession session = borrowMessengerSession();

         try {

             MessageConsumer consumer = createMessageConsumer(session, session.getListenerSession(), destination);

             consumer.setMessageListener(listener);

             ListenerKey key = new ListenerKey(destination, listener);

             listeners.put(key, consumer);

         }

         finally {

             returnMessengerSession(session);

         }

     }

     public void addListener(Destination destination, String selector, MessageListener listener) throws JMSException {

         if (listener instanceof MessengerListener) {

             MessengerListener messengerListener = (MessengerListener) listener;

             messengerListener.setMessenger(this);

         }

         MessengerSession session = borrowMessengerSession();

         try {

             MessageConsumer consumer = createMessageConsumer(session, session.getListenerSession(), destination, selector);

             consumer.setMessageListener(listener);

             ListenerKey key = new ListenerKey(destination, listener, selector);

             listeners.put(key, consumer);

         }

         finally {

             returnMessengerSession(session);

         }

     }

     public void removeListener(Destination destination, MessageListener listener) throws JMSException {

         ListenerKey key = new ListenerKey(destination, listener);

         MessageConsumer consumer = (MessageConsumer) listeners.remove(key);

         if (consumer == null) {

             throw new JMSException("The given listener object has not been added for the given destination");

         }

         consumer.close();

     }

     public void removeListener(Destination destination, String selector, MessageListener listener)

         throws JMSException {

         ListenerKey key = new ListenerKey(destination, listener, selector);

         MessageConsumer consumer = (MessageConsumer) listeners.remove(key);

         if (consumer == null) {

             throw new JMSException("The given listener object has not been added for the given destination and selector");

         }

         consumer.close();

     }

     // Message factory methods

     //-------------------------------------------------------------------------

     public BytesMessage createBytesMessage() throws JMSException {

         MessengerSession session = borrowMessengerSession();

         try {

             return session.getSession().createBytesMessage();

         }

         finally {

             returnMessengerSession(session);

         }

     }

     public MapMessage createMapMessage() throws JMSException {

         MessengerSession session = borrowMessengerSession();

         try {

             return session.getSession().createMapMessage();

         }

         finally {

             returnMessengerSession(session);

         }

     }

     public Message createMessage() throws JMSException {

         MessengerSession session = borrowMessengerSession();

         try {

             return session.getSession().createMessage();

         }

         finally {

             returnMessengerSession(session);

         }

     }

     public ObjectMessage createObjectMessage() throws JMSException {

         MessengerSession session = borrowMessengerSession();

         try {

             return session.getSession().createObjectMessage();

         }

         finally {

             returnMessengerSession(session);

         }

     }

     public ObjectMessage createObjectMessage(Serializable object) throws JMSException {

         MessengerSession session = borrowMessengerSession();

         try {

             return session.getSession().createObjectMessage(object);

         }

         finally {

             returnMessengerSession(session);

         }

     }

     public StreamMessage createStreamMessage() throws JMSException {

         MessengerSession session = borrowMessengerSession();

         try {

             return session.getSession().createStreamMessage();

         }

         finally {

             returnMessengerSession(session);

         }

     }

     public TextMessage createTextMessage() throws JMSException {

         MessengerSession session = borrowMessengerSession();

         try {

             return session.getSession().createTextMessage();

         }

         finally {

             returnMessengerSession(session);

         }

     }

     public TextMessage createTextMessage(String text) throws JMSException {

         MessengerSession session = borrowMessengerSession();

         try {

             return session.getSession().createTextMessage(text);

         }

         finally {

             returnMessengerSession(session);

         }

     }

     public void commit() throws JMSException {

         MessengerSession session = borrowMessengerSession();

         try {

             session.getSession().commit();

         }

         finally {

             returnMessengerSession(session);

         }

     }

     public void rollback() throws JMSException {

         MessengerSession session = borrowMessengerSession();

         try {

             session.getSession().rollback();

         }

         finally {

             returnMessengerSession(session);

         }

     }

     public void close() throws JMSException {

         getSessionFactory().close();

     }

     /**

      * Creates a browser on the given Queue

      */

     public QueueBrowser createBrowser(Destination destination) throws JMSException {

         MessengerSession session = borrowMessengerSession();

         QueueBrowser browser = null;

         try {

             return createBrowser(session, destination);

         }

         finally {

             returnMessengerSession(session);

         }

     }

     /** Get the producer's default delivery mode. */

     public int getDeliveryMode(Destination destination) throws JMSException {

         MessengerSession session = borrowMessengerSession();

         int deliveryMode = 0;

         try {

             MessageProducer producer = session.getMessageProducer(destination);

             deliveryMode = producer.getDeliveryMode();

         }

         finally {

             returnMessengerSession(session);

         }

         return deliveryMode;

     }

     /** Set the producer's default delivery mode. */

     public void setDeliveryMode(Destination destination, int deliveryMode) throws JMSException {

         MessengerSession session = borrowMessengerSession();

         MessageProducer producer = null;

         try {

             producer = session.getMessageProducer(destination);

             producer.setDeliveryMode(deliveryMode);

         }

         finally {

             returnMessengerSession(session);

         }

     }

     /**  Get the producer's default priority. */

     public int getPriority(Destination destination) throws JMSException {

         MessengerSession session = borrowMessengerSession();

         MessageProducer producer = null;

         int priority = 0;

         try {

             producer = session.getMessageProducer(destination);

             priority = producer.getPriority();

         }

         finally {

             returnMessengerSession(session);

         }

         return priority;

     }

     /**   Set the producer's default priority. */

     public void setPriority(Destination destination, int priority) throws JMSException {

         MessengerSession session = borrowMessengerSession();

         MessageProducer producer = null;

         try {

             producer = session.getMessageProducer(destination);

             producer.setPriority(priority);

         }

         finally {

             returnMessengerSession(session);

         }

     }

     /**  Get the producer's default delivery mode. */

     public long getTimeToLive(Destination destination) throws JMSException {

         MessengerSession session = borrowMessengerSession();

         long timeToLive = 0;

         try {

             MessageProducer producer = session.getMessageProducer(destination);

             timeToLive = producer.getTimeToLive();

         }

         finally {

             returnMessengerSession(session);

         }

         return timeToLive;

     }

     /**  <p>Set the default length of time in milliseconds from its dispatch time that

      *   a produced message should be retained by the message system.</p>

      */

     public void setTimeToLive(Destination destination, long timeToLive) throws JMSException {

         MessengerSession session = borrowMessengerSession();

         try {

             MessageProducer producer = session.getMessageProducer(destination);

             producer.setTimeToLive(timeToLive);

         }

         finally {

             returnMessengerSession(session);

         }

     }

     /** Get an indication of whether message timestamps are disabled. */

     public boolean getDisableMessageTimestamp(Destination destination) throws JMSException {

         MessengerSession session = borrowMessengerSession();

         boolean value = false;

         try {

             MessageProducer producer = session.getMessageProducer(destination);

             value = producer.getDisableMessageTimestamp();

         }

         finally {

             returnMessengerSession(session);

         }

         return value;

     }

     /** Set whether message timestamps are disabled. */

     public void setDisableMessageTimestamp(Destination destination, boolean value) throws JMSException {

         MessengerSession session = borrowMessengerSession();

         try {

             MessageProducer producer = session.getMessageProducer(destination);

             producer.setDisableMessageTimestamp(value);

         }

         finally {

             returnMessengerSession(session);

         }

     }

     /** Extends the send capability to send by specifying additional options. */

     public void send(Destination destination, Message message, int deliveryMode, int priority, long timeToLive)

         throws JMSException {

         MessengerSession session = borrowMessengerSession();

         try {

             MessageProducer producer = session.getMessageProducer(destination);

             if (session.isTopic()) {

                 ((TopicPublisher) producer).publish((Topic) destination, message, deliveryMode, priority, timeToLive);

             }

             else {

                 ((QueueSender) producer).send((Queue) destination, message, deliveryMode, priority, timeToLive);

             }

         }

         finally {

             returnMessengerSession(session);

         }

     }

     /**  Get an indication of whether message IDs are disabled. */

     public boolean getDisableMessageID(Destination destination) throws JMSException {

         MessengerSession session = borrowMessengerSession();

         boolean value = false;

         try {

             MessageProducer producer = session.getMessageProducer(destination);

             value = producer.getDisableMessageID();

         }

         finally {

             returnMessengerSession(session);

         }

         return value;

     }

     /** Set whether message IDs are disabled. */

     public void setDisableMessageID(Destination destination, boolean value) throws JMSException {

         MessengerSession session = borrowMessengerSession();

         try {

             MessageProducer producer = session.getMessageProducer(destination);

             producer.setDisableMessageID(value);

         }

         finally {

             returnMessengerSession(session);

         }

     }

     // Properties

     //-------------------------------------------------------------------------

     /** Gets the name that this Messenger is called in a MessengerManager */

     public String getName() {

         return name;

     }

     /** Sets the name that this Messenger is called in a MessengerManager */

     public void setName(String name) {

         this.name = name;

     }

     /** Setter for jndiDestinations */

     public void setJndiDestinations(boolean jndiDestinations) {

         this.jndiDestinations = jndiDestinations;

     }

     /** Getter for jndiDestinations */

     public boolean isJndiDestinations() {

         return jndiDestinations;

     }

     /** Gets whether topic subscribers are durable or not */

     public boolean isDurable() {

         return durable;

     }

     /** Sets whether topic subscribers are durable or not */

     public void setDurable(boolean durable) {

         this.durable = durable;

     }

     /** Gets whether we should cache the requestor object per thread? */

     public boolean isCacheRequestors() {

         return cacheRequestors;

     }

     /** Sets whether we should cache the requestor object per thread? */

     public void setCacheRequestors(boolean cacheRequestors) {

         this.cacheRequestors = cacheRequestors;

     }

     /** @return the durable name used for durable topic based subscriptions */

     public String getDurableName() {

         return durableName;

     }

     /** Sets the durable name used for durable topic based subscriptions */

     public void setDurableName(String durableName) {

         this.durableName = durableName;

     }

     /** 

      * Gets whether local messages are ignored when topic based subscription is used

      * with a message selector 

      */

     public boolean isNoLocal() {

         return noLocal;

     }

     /** 

      * Sets whether local messages are ignored when topic based subscription is used

      * with a message selector 

      */

     public void setNoLocal(boolean noLocal) {

         this.noLocal = noLocal;

     }

     /** Gets whether MessageProducer instances should be cached or not, which defaults to true */

     public boolean isCacheProducers() {

         return cacheProducers;

     }

     /** Sets whether MessageProducer instances should be cached or not, which defaults to true */

     public void setCacheProducers(boolean cacheProducers) {

         this.cacheProducers = cacheProducers;

     }

     /**

      * Returns the delivery mode used on messages sent via this Messenger

      * @return int

      */

     public int getDeliveryMode() {

         return deliveryMode;

     }

     /**

      * Sets the delivery mode used on messages sent via this Messenger

      * @param deliveryMode The deliveryMode to set

      */

     public void setDeliveryMode(int deliveryMode) {

         this.deliveryMode = deliveryMode;

     }

     /**

      * Sets whether message delivery should be persistent or not

      * 

      * @param persistentDelivery

      */

     public void setPersistentDelivery(boolean persistentDelivery) {

         if (persistentDelivery) {

             setDeliveryMode(DeliveryMode.PERSISTENT);

         }

         else {

             setDeliveryMode(DeliveryMode.NON_PERSISTENT);

         }

     }

     // Implementation methods

     //-------------------------------------------------------------------------

     protected abstract MessengerSession borrowMessengerSession() throws JMSException;

     protected abstract void returnMessengerSession(MessengerSession session);

     protected abstract boolean isTopic(Connection connection) throws JMSException;

     protected abstract boolean isTopic(ConnectionFactory factory) throws JMSException;

     /** @return a newly created message producer for the given session and destination */

     protected MessageProducer createMessageProducer(MessengerSession messengerSession, Destination destination) throws JMSException {

         MessageProducer answer = null;

         Session session = messengerSession.getSession();

         if (messengerSession.isTopic()) {

             TopicSession topicSession = (TopicSession) session;

             answer = topicSession.createPublisher((Topic) destination);

         }

         else {

             QueueSession queueSession = (QueueSession) session;

             answer = queueSession.createSender((Queue) destination);

         }

         // configure the MessageProducer

         if (deliveryMode != Message.DEFAULT_DELIVERY_MODE) {

             answer.setDeliveryMode(deliveryMode);

         }

         return answer;

     }

     /** @return a MessageConsumer for the given session and destination */

     protected MessageConsumer borrowMessageConsumer(MessengerSession messengerSession, Session session, Destination destination)

         throws JMSException {

         MessageConsumer consumer = createMessageConsumer(messengerSession, session, destination);

         if (log.isDebugEnabled()) {

             log.debug("Created new consumer: " + consumer + " on destination: " + destination);

         }

         return consumer;

     }

     /** @return a MessageConsumer for the given session, destination and selector */

     protected MessageConsumer borrowMessageConsumer(MessengerSession messengerSession, Session session, Destination destination, String selector)

         throws JMSException {

         MessageConsumer consumer = createMessageConsumer(messengerSession, session, destination, selector);

         if (log.isDebugEnabled()) {

             log.debug(

                 "Created new consumer: " + consumer + " on destination: " + destination + " selector: " + selector);

         }

         return consumer;

     }

     /** 

      * Returns a message consumer back to the pool. 

      * By default this method will close message consumers though we should

      * be able to cache then

      */

     protected void returnMessageConsumer(MessageConsumer messageConsumer) throws JMSException {

         if (log.isDebugEnabled()) {

             log.debug("Closing consumer: " + messageConsumer);

         }

         if (messageConsumer != null) {

             messageConsumer.close();

         }

     }

     /** @return a new MessageConsumer for the given session and destination */

     protected MessageConsumer createMessageConsumer(MessengerSession messengerSession, Session session, Destination destination)

         throws JMSException {

         if (messengerSession.isTopic()) {

             TopicSession topicSession = (TopicSession) session;

             if (isDurable()) {

                 return topicSession.createDurableSubscriber((Topic) destination, getDurableName(), null, isNoLocal());

             }

             else {

                 return topicSession.createSubscriber((Topic) destination, null, isNoLocal());

             }

         }

         else {

             QueueSession queueSession = (QueueSession) session;

             return queueSession.createReceiver((Queue) destination);

         }

     }

     /** @return a new MessageConsumer for the given session, destination and selector */

     protected MessageConsumer createMessageConsumer(

         MessengerSession messengerSession,

         Session session,

         Destination destination,

         String selector)

         throws JMSException {

         if (messengerSession.isTopic()) {

             TopicSession topicSession = (TopicSession) session;

             if (isDurable()) {

                 return topicSession.createDurableSubscriber(

                     (Topic) destination,

                     getDurableName(),

                     selector,

                     isNoLocal());

             }

             else {

                 return topicSession.createSubscriber((Topic) destination, selector, isNoLocal());

             }

         }

         else {

             QueueSession queueSession = (QueueSession) session;

             return queueSession.createReceiver((Queue) destination, selector);

         }

     }

     /** @return a new QueueBrowser for the given session and destination */

     protected QueueBrowser createBrowser(MessengerSession session, Destination destination) throws JMSException {

         if (session.isTopic()) {

             return null;

         }

         else {

             QueueSession queueSession = (QueueSession) session.getSession();

             return queueSession.createBrowser((Queue) destination);

         }

     }

     protected Queue getQueue(QueueSession session, String subject) throws JMSException {

         // XXXX: might want to cache

         return session.createQueue(subject);

     }

     protected Topic getTopic(TopicSession session, String subject) throws JMSException {

         // XXXX: might want to cache

         return session.createTopic(subject);

     }

 }