JmsManagerImpl.java

/*
 * JOnAS: Java(TM) Open Application Server
 * Copyright (C) 1999 Bull S.A.
 * Contact: jonas-team@objectweb.org
 * 
 * This library is free software; you can redistribute it and/or
 * modify it under the terms of the GNU Lesser General Public
 * License as published by the Free Software Foundation; either
 * version 2.1 of the License, or any later version.
 * 
 * This library is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
 * Lesser General Public License for more details.
 * 
 * You should have received a copy of the GNU Lesser General Public
 * License along with this library; if not, write to the Free Software
 * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307
 * USA
 *
 * Initial developer(s): 
 * Contributor(s): 
 * Christophe Ney: for making easier Enhydra integration
 * Philippe Durieux
 * Jeff Mesnil: for JORAM 3.0 integration
 * --------------------------------------------------------------------------
 * $Id: JmsManagerImpl.java,v 1.28 2004/10/06 14:41:54 benoitf Exp $
 * --------------------------------------------------------------------------
 */


package org.objectweb.jonas_jms;

import java.util.Enumeration;
import java.util.Hashtable;
import java.util.Vector;

import javax.jms.ConnectionFactory;
import javax.jms.JMSException;
import javax.jms.Queue;
import javax.jms.QueueConnectionFactory;
import javax.jms.Topic;
import javax.jms.TopicConnectionFactory;
import javax.jms.XAConnectionFactory;
import javax.jms.XAQueueConnectionFactory;
import javax.jms.XATopicConnectionFactory;
import javax.naming.InitialContext;
import javax.naming.NamingException;

import org.objectweb.jonas_jms.api.JmsAdministration;
import org.objectweb.jonas_jms.api.JmsManager;
import org.objectweb.transaction.jta.TransactionManager;
import org.objectweb.util.monolog.api.BasicLevel;

public class JmsManagerImpl implements JmsManager, JmsJmxManagement {

    private JmsAdministration momadmin = null;
    private InitialContext ictx = null;
    private ConnectionFactory cf = null;
    private TopicConnectionFactory tcf = null;
    private QueueConnectionFactory qcf = null;
    private Hashtable queues = new Hashtable();
    private Hashtable topics = new Hashtable();
    private Vector namelist = new Vector();

    private static TransactionManager tm = null;
    private static JmsManagerImpl unique = null;

    private JmsManagerImpl() {
        TraceJms.logger.log(BasicLevel.DEBUG,"");
    }

    public static JmsManager getJmsManager() {
        if (unique == null) {
            unique = new JmsManagerImpl();
        }
        return (JmsManager) unique;
    }

    public static JmsJmxManagement getJmsJmxManagement() {
        if (unique == null) {
            unique = new JmsManagerImpl();
        }
        return (JmsJmxManagement) unique;
    }

    // -------------------------------------------------------------------
    // Internal Methods
    // -------------------------------------------------------------------

    public static TransactionManager getTransactionManager() {
        return tm;
    }

    // -------------------------------------------------------------------
    // JmsManager  Implementation
    // -------------------------------------------------------------------

    public void init(Class cl, boolean collocated, String url, TransactionManager trm) throws Exception  {
        TraceJms.logger.log(BasicLevel.DEBUG,"");
        tm = trm;
        int maxloops = collocated ? 1 : 5;
        for (int i = 0; i < maxloops; i++) {
            try {           
                // Create an InitialContext
                ictx = new InitialContext();
                // Create the MOM instance and start it
                momadmin  = (JmsAdministration) cl.newInstance();
                momadmin.start(collocated, url);
                break;
            } catch (NamingException e) {
                TraceJms.logger.log(BasicLevel.ERROR, "cannot get Initial Context", e);
                throw e;
            } catch (NullPointerException e) {
                TraceJms.logger.log(BasicLevel.ERROR, "exception: ", e);
                throw e;
            } catch (Exception e) {
                if (i < maxloops) {
                    if(TraceJms.isDebug())
                        TraceJms.logger.log(BasicLevel.DEBUG, "cannot reach the MOM - retrying...");
                    try {
                        // Thread.sleep() -> force current Thread to sleep
                        Thread.sleep(2000*(i+1));
                    } catch (InterruptedException e2) {
                        throw new JMSException("Cannot reach the MOM");
                    }
                } else {
                    TraceJms.logger.log(BasicLevel.ERROR, "cannot load admin class "+e);
                    throw e;
                }
            }
        }
        // We must register these factories in JNDI before any client 
        // try to get them.
        getConnectionFactory();
        getTopicConnectionFactory();
        getQueueConnectionFactory();
    }   

    public Queue createQueue(String name) throws Exception {
        Queue queue  = null;
        // Don't recreate if already found in JNDI
        try {
            queue = (Queue) ictx.lookup(name);
            if(TraceJms.isDebug())
                TraceJms.logger.log(BasicLevel.DEBUG, "queue " + name +" already found");
            return queue;
        } catch (NamingException ex) {
        }
        if(TraceJms.isDebug())
            TraceJms.logger.log(BasicLevel.DEBUG, "creating and registering queue "+name);
        queue = momadmin.createQueue(name);
        // rebind is already done in momadmin.createQueue()
        //ictx.rebind(name, queue);
        namelist.addElement(name);
        queues.put(name, queue);
        return queue;
    }
        
    public Topic createTopic(String name) throws Exception {
        Topic topic = null;
        // Don't recreate if already found in JNDI
        try {
            topic = (Topic) ictx.lookup(name);
            if(TraceJms.isDebug())
                TraceJms.logger.log(BasicLevel.DEBUG, "topic " + name + " already found");
            return topic;
        } catch (NamingException ex) {
        }
        if(TraceJms.isDebug())
            TraceJms.logger.log(BasicLevel.DEBUG, "creating and registering topic " + name);
        topic = momadmin.createTopic(name);
        // rebind is already done in momadmin.createTopic()
        //ictx.rebind(name, topic);
        namelist.addElement(name);
        topics.put(name, topic);
        return topic;
    }

    public ConnectionFactory getConnectionFactory() {
        if (cf == null) {
            String name = "CF";
            cf = new org.objectweb.jonas_jms.JConnectionFactory(name);
            try {
                if(TraceJms.isDebug())
                    TraceJms.logger.log(BasicLevel.DEBUG, "creating and registering "+name);
                ictx.rebind(name, cf);
            } catch (NamingException e) {
                TraceJms.logger.log(BasicLevel.ERROR, "cannot rebind "+name+" :"+e);
            }
        }
        return cf;
    }

    public TopicConnectionFactory getTopicConnectionFactory() {
        if (tcf == null) {
            String name = "TCF";
            tcf = new org.objectweb.jonas_jms.JTopicConnectionFactory(name);
            try {
                if(TraceJms.isDebug())
                    TraceJms.logger.log(BasicLevel.DEBUG,"creating and registering " + name);
                ictx.rebind(name, tcf);
            } catch (NamingException e) {
                TraceJms.logger.log(BasicLevel.ERROR, "cannot rebind " + name + " :" + e);
            }
        }
        return tcf;
    }
 
    /*
     *  Get the unique QueueConnectionFactory
     */
    public QueueConnectionFactory getQueueConnectionFactory() {
        if (qcf == null) {
            String name = "QCF";
            qcf = new org.objectweb.jonas_jms.JQueueConnectionFactory(name);
            try {
                if(TraceJms.isDebug())
                    TraceJms.logger.log(BasicLevel.DEBUG, "creating and registering " + name);
                ictx.rebind(name, qcf);
            } catch (NamingException e) {
                TraceJms.logger.log(BasicLevel.ERROR, "cannot rebind " + name + " :" + e);
            }
        }
        return qcf;
    }

    public XAConnectionFactory getXAConnectionFactory() {
        return momadmin.getXAConnectionFactory();
    }

    public XATopicConnectionFactory getXATopicConnectionFactory() {
        return momadmin.getXATopicConnectionFactory();
    }
 
    public XAQueueConnectionFactory getXAQueueConnectionFactory() {
        return momadmin.getXAQueueConnectionFactory();
    }

    public Queue getQueue(String name) throws Exception {
        Queue q = (Queue)queues.get(name);
        if (q == null) {
            q = createQueue(name);
        }
        return q;
    }

    public Topic getTopic(String name) throws Exception {
        Topic t = (Topic)topics.get(name);
        if (t == null) {
            t = createTopic(name);
        }
        return t;
    }

    public Enumeration getTopicsNames() {
        return topics.keys();
    }   

    public Enumeration getQueuesNames() {
        return queues.keys();
    }

    public void stop() throws Exception {
        // Before stopping the MOM clean up the connection pools
        if (cf != null) {
            ((JConnectionFactory) cf).cleanPool();
        }

        if (tcf != null) {
            ((JConnectionFactory) tcf).cleanPool();
        }

        if (qcf != null) {
            ((JConnectionFactory) qcf).cleanPool();
        }

        // Stop the MOM
        if (momadmin != null) {
            momadmin.stop();
        }

        // clean up JNDI        
        Enumeration ln = namelist.elements();
        while (ln.hasMoreElements()) {
            String name = (String)ln.nextElement();
            try {                   
                ictx.unbind(name);              
                if(TraceJms.isDebug())
                    TraceJms.logger.log(BasicLevel.DEBUG, "unbind " + name);
            } catch(NamingException e) {
                if(TraceJms.isDebug())
                    TraceJms.logger.log(BasicLevel.ERROR, "cannot unbind " + name);
            }
        }
        
    }

    // -------------------------------------------------------------
    // Management methods
    // -------------------------------------------------------------

    public int getCurrentNumberOfJmsConnectionFactory() {
        int result = 0;
        if (cf != null) result++;
        return result;
    }

    public int getCurrentNumberOfJmsTopicConnectionFactory() {
        int result = 0;
        if (tcf != null) result++;
        return result;
    }

    public int getCurrentNumberOfJmsQueueConnectionFactory() {
        int result = 0;
        if (qcf != null) result++;
        return result;
    }

    public int getCurrentNumberOfJmsTopicDestination() {
        return topics.size();
    }

    public int getCurrentNumberOfJmsQueueDestination() {
        return queues.size();
    }

    public String removeJmsDestination(String jndiName) throws Exception {

        momadmin.deleteDestination(jndiName);

        String destType = null;
        if ((queues.containsKey(jndiName))||(topics.containsKey(jndiName))) {
            try {           
                InitialContext ictx = new InitialContext();
                ictx.unbind(jndiName);
                Object o = null; // object to which the jndiName is mapped
                o = queues.remove(jndiName);
                if (o == null) {
                    // jndiName did not have a mapping, try in the topics hashtable
                    o = topics.remove(jndiName);
                    if (o != null) {
                        destType = "topic"; 
                    } else {
                        // nither queue nor topic ??? 
                        destType = "unknown";
                    }
                } else {
                    destType = "queue";
                }
                return destType;
            } catch(Exception ex) {
                throw new Exception("JmsManagerImpl remove destination : cannot unbind " + jndiName );
            }
        } else {
            throw new Exception("Unexisting jms destination :" + jndiName +" remove abord");
        }
    }

    public String getDefaultConnectionFactoryName() {
        return "CF"; 
    }
    
    public String getDefaultQueueConnectionFactoryName() {
        return "QCF"; 
    }

    public String getDefaultTopicConnectionFactoryName() {
        return "TCF"; 
    }

    // Monitoring methods

    public String getConnectionFactoryMode(String jndiName) throws Exception {
        TraceJms.logger.log(BasicLevel.DEBUG, "");

        if (jndiName.equals("CF"))
            return "Point-To-Point and Publish/Subscribe";
        else if (jndiName.equals("QCF"))
            return "Point-To-Point";
        else if (jndiName.equals("TCF"))
            return "Publish/Subscribe";
        else
            throw new IllegalStateException("Unknown factory.");
    }

 
    public int getPendingMessages(String jndiName) throws Exception {
        TraceJms.logger.log(BasicLevel.DEBUG, "");
        Queue queue = (Queue)queues.get(jndiName);
        int n;
        if (queue != null) {
            // existent queue (created by JOnAS)
            n = momadmin.getPendingMessages(queue);
        } else {
            throw new IllegalStateException(jndiName + " not a queue created by the jms service");
        }
        return n;
    }

    public int getPendingRequests(String jndiName) throws Exception {
        TraceJms.logger.log(BasicLevel.DEBUG, "");
        Queue queue = (Queue)queues.get(jndiName);
        int n;
        if (queue != null) {
            // existent queue (created by JOnAS)
            n = momadmin.getPendingRequests(queue);
        } else {
            throw new IllegalStateException(jndiName + " not a queue created by the jms service");
        }
        return n;
    } 

    public int getSubscriptions(String jndiName) throws Exception {
        Topic topic = (Topic)topics.get(jndiName);
        int n;
        if (topic != null) {
            // existent topic (created by JOnAS)
            n = momadmin.getSubscriptions(topic);
        } else {
            throw new IllegalStateException(jndiName + " not a topic created by the jms service");
        }
        return n;
    }  
}

---