JmsServiceImpl.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
 * --------------------------------------------------------------------------
 * $Id: JmsServiceImpl.java,v 1.28 2004/03/25 15:58:20 sauthieg Exp $
 * --------------------------------------------------------------------------
 */


package org.objectweb.jonas.jms;

import java.util.Enumeration;
import java.util.HashSet;
import java.util.Set;
import java.util.StringTokenizer;
import java.util.Vector;

import javax.jms.Queue;
import javax.jms.Topic;
import javax.naming.Context;
import javax.naming.NamingException;

import org.objectweb.jonas.common.Log;
import org.objectweb.jonas.jmx.JmxService;
import org.objectweb.jonas.jmx.JonasObjectName;
import org.objectweb.jonas.jtm.TransactionService;
import org.objectweb.jonas.management.ReconfiguredProp;
import org.objectweb.jonas.service.AbsServiceImpl;
import org.objectweb.jonas.service.ServiceException;
import org.objectweb.jonas.service.ServiceManager;
import org.objectweb.jonas_jms.JmsJmxManagement;
import org.objectweb.jonas_jms.JmsManagerImpl;
import org.objectweb.jonas_jms.api.JmsManager;
import org.objectweb.util.monolog.api.BasicLevel;
import org.objectweb.util.monolog.api.Logger;

public class JmsServiceImpl extends AbsServiceImpl implements JmsService, JmsServiceImplMBean {

    // loggers
    static private Logger serverlog = null;
    static private Logger loaderlog = null;

    // Service name as used to label configuration properties
    public static final String SERVICE_NAME = "jms";

    // Configuration information used when starting the JMS service
    String momProviderClassName;
    Vector queueNames = new Vector();
    Vector topicNames = new Vector();
    boolean isMomLocal;
    String url;
    String jmsClass;
    TransactionService transactionService = null;
    
    // JMS manager
    JmsManager jonasjms = null;
    JmsJmxManagement jmsjmx = null;

    // JMS service configuration properties
    public static final String COLLOCATED = "jonas.service.jms.collocated";
    public static final String MOM = "jonas.service.jms.mom";
    public static final String TOPICS = "jonas.service.jms.topics";
    public static final String QUEUES = "jonas.service.jms.queues";
    public static final String DESTINATIONS = "jonas.service.jms.dest";
    public static final String URL = "jonas.service.jms.url";
    public static final String CLASS = "jonas.service.jms.class";

    // Value used as sequence number by reconfiguration notifications
    long sequenceNumber = 0;

    // -------------------------------------------------------------------
    // Service Implementation
    // -------------------------------------------------------------------

    public void doInit(Context ctx) throws ServiceException {

        // get loggers
        serverlog = Log.getLogger(Log.JONAS_SERVER_PREFIX);
        loaderlog = Log.getLogger(Log.JONAS_LOADER_PREFIX);
    super.initLogger(Log.getLogger(Log.JONAS_MANAGEMENT_PREFIX));

        // Read the configuration information
        try {
            momProviderClassName = (String) ctx.lookup(MOM);
        } catch (NamingException e) {
            serverlog.log(BasicLevel.ERROR, "JMS Service Cannot read configuration "+e);
            throw new ServiceException("JMS Service Cannot read configuration ", e);
        }

        String ql = null;
        try {
            ql = (String) ctx.lookup(QUEUES);
        } catch (NamingException e) {
            // No queue predefined
        }
        if (ql != null) {
            StringTokenizer st = new StringTokenizer(ql, ",");
            while (st.hasMoreTokens()) {
                queueNames.add(st.nextToken().trim());
            }
        }

        String tl = null;
        try {
            tl = (String) ctx.lookup(TOPICS);
        } catch (NamingException e) {
            // No topic predefined
        }
        if (tl != null) {
            StringTokenizer st = new StringTokenizer(tl, ",");
            while (st.hasMoreTokens()) {
                topicNames.add(st.nextToken().trim());
            }
        }

        String local = "true";
        try {
            local = (String) ctx.lookup(COLLOCATED);
        } catch (NamingException e) {
            // default value is collocated
        }
        isMomLocal = local.equalsIgnoreCase("true");

        url = "";
        try {
            url = (String) ctx.lookup(URL);
        } catch (NamingException e) {
            // default value is null string
        }

        // Get the transaction service
        try {
            transactionService =
                (TransactionService) ServiceManager.getInstance().getTransactionService();
        } catch (Exception e) {
            serverlog.log(BasicLevel.ERROR, "Error when starting the JMS service "+e);
            throw new ServiceException("Error when starting the JMS service ", e);
        }

        serverlog.log(BasicLevel.DEBUG, "JMS service initialized");
    }

    public void doStart() throws ServiceException {

        if (isMomLocal) {
            serverlog.log(BasicLevel.DEBUG, "Starting JMS Service with collocated MOM");
        } else {
            serverlog.log(BasicLevel.DEBUG, "Starting JMS Service with MOM at url "+url);
        }
        
        jonasjms = JmsManagerImpl.getJmsManager();
        jmsjmx = JmsManagerImpl.getJmsJmxManagement();
        
        Class c = null;
        loaderlog.log(BasicLevel.DEBUG, "JmsServiceImpl classloader="+getClass().getClassLoader());
        try {
            ClassLoader loader = JmsManagerImpl.class.getClassLoader();
            if (loader == null) {
                loader = Thread.currentThread().getContextClassLoader();
            }
            c = loader.loadClass(momProviderClassName);
            loaderlog.log(BasicLevel.DEBUG, momProviderClassName+" classloader="+loader);
        } catch (Exception e) {
            serverlog.log(BasicLevel.ERROR, "JMS Service Cannot load admin class "+e);
            throw new ServiceException("JMS Service Cannot load admin class", e);
        }
        
        // initialization of the JmsManager with implementation class for Administration
        try {
            jonasjms.init(c, isMomLocal, url, transactionService.getTransactionManager());
        } catch (Exception e) {
            serverlog.log(BasicLevel.ERROR, "JMS Service Cannot init the JMS manager "+e);
            throw new ServiceException("JMS Service Cannot init the JMS manager", e);
        }
        
        // Create Topics and Queues
        try {
            // Create administered objects for Queues
            for (int i = 0; i < queueNames.size(); i++) {
                Queue q = jonasjms.createQueue((String) queueNames.elementAt(i));
            }

            // Create administered objects for Topics
            for (int i = 0; i < topicNames.size(); i++) {
                Topic t = jonasjms.createTopic((String) topicNames.elementAt(i));
            }
        } catch (Exception ex) {
            serverlog.log(BasicLevel.ERROR, "JMS Service Cannot create administered object : "+ex.toString());
            try {
                jonasjms.stop();
                jonasjms = null;
            } catch (Exception e) {
            }
            throw new ServiceException("JMS Service Cannot create administered object", ex);
        }

        try {
            // Register JmsService MBean : JmxJmsService 
            ((JmxService)ServiceManager.getInstance().getJmxService()).getJmxServer().registerMBean(this,JonasObjectName.jmsService());
        } catch (ServiceException se) {
            // Jmx Service not available, do nothing
        } catch (Exception e) {
            serverlog.log(BasicLevel.ERROR, "JmsService: Cannot start the JMS service:\n"+e);
            throw new ServiceException("JmsService: Cannot start the JMS service",
                                       e);
        }

    }

    public void doStop() throws ServiceException {

        try {
            jonasjms.stop();
        } catch (Exception e) {
            serverlog.log(BasicLevel.ERROR, "JMS Service Cannot stop the JMS manager "+e);
            throw new ServiceException("JMS Service Cannot stop the JMS manager", e);
        }
        // unregister MBean     
        try {
            // unregister jms Service MBean
            ((JmxService)ServiceManager.getInstance().getJmxService()).getJmxServer().unregisterMBean(JonasObjectName.jmsService());
        } catch (ServiceException se) {
            // Jmx Service not available, do nothing
        } catch (Exception e) {
            serverlog.log(BasicLevel.ERROR, "Cannot stop the JMS service:\n"+e);
            throw new ServiceException("Cannot stop the JMS service",e);
        }
        serverlog.log(BasicLevel.DEBUG, "JMS Service stopped");
    }


    // -------------------------------------------------------------------
    // JMS Service Implementation
    // -------------------------------------------------------------------

    public JmsManager  getJmsManager() {
        return jonasjms;
    }

    public Integer getCurrentNumberOfJmsConnectionFactory(){
            return new Integer(jmsjmx.getCurrentNumberOfJmsConnectionFactory());
    }

    public Integer getCurrentNumberOfJmsTopicConnectionFactory(){
        return new Integer(jmsjmx.getCurrentNumberOfJmsTopicConnectionFactory());

    }

    public Integer getCurrentNumberOfJmsQueueConnectionFactory(){
                return new Integer(jmsjmx.getCurrentNumberOfJmsQueueConnectionFactory());
    }

    public Integer getCurrentNumberOfJmsTopicDestination(){
        return new Integer(jmsjmx.getCurrentNumberOfJmsTopicDestination());
    }

    public Integer getCurrentNumberOfJmsQueueDestination(){
        return new Integer(jmsjmx.getCurrentNumberOfJmsQueueDestination());
    }

    public void createJmsQueueDestination(String jndiName){
        try {
            jonasjms.createQueue(jndiName);
            
        // Send to the listner MBean a notification containing the new value of this property and 
        // the indication that the new value has to be added to a list of values 
        ReconfiguredProp p = new ReconfiguredProp(QUEUES, jndiName, false);
        p.setAdd(true); 
        sendReconfigNotification(++sequenceNumber, SERVICE_NAME, p);
        } catch (Exception e) {
            serverlog.log(BasicLevel.ERROR, "JmsService: Exception when create destination " + jndiName + ": " + e);
        }
    } 

    public void createJmsTopicDestination(String jndiName){
        try {
            jonasjms.createTopic(jndiName);

            // Send to the listner MBean a notification containing the new value of this property and 
            // the indication that the new value has to be added to a list of values 
            ReconfiguredProp p = new ReconfiguredProp(TOPICS, jndiName, false);
            p.setAdd(true);
            sendReconfigNotification(++sequenceNumber, SERVICE_NAME, p);
        } catch (Exception e) {
            serverlog.log(BasicLevel.ERROR, "JmsService: Exception when create destination " + jndiName + ": " + e);
        }   
    } 

    public void removeJmsTopicDestination(String jndiName){
        try{
            jmsjmx.removeJmsDestination(jndiName);

            // Send to the listner MBean a notification containing the new value of this property and 
            // the indication that the new value has to be removed from a list of values 
            ReconfiguredProp p = new ReconfiguredProp(TOPICS, jndiName, false);
            p.setAdd(false);
            sendReconfigNotification(++sequenceNumber, SERVICE_NAME, p);
        } catch (Exception e){
            // remove problem
            serverlog.log(BasicLevel.ERROR, "JmsService: Exception when remove destination"+e);
        }
    }

    public void removeJmsQueueDestination(String jndiName){
        try{
            jmsjmx.removeJmsDestination(jndiName);

            // Send to the listner MBean a notification containing the new value of this property and 
            // the indication that the new value has to be removed from a list of values 
            ReconfiguredProp p = new ReconfiguredProp(QUEUES, jndiName, false);
            p.setAdd(false);
            sendReconfigNotification(++sequenceNumber, SERVICE_NAME, p);
        } catch (Exception e){
            // remove problem
            serverlog.log(BasicLevel.ERROR, "JmsService: Exception when remove destination"+e);
        }
    }


    public void removeJmsDestination(String jndiName) {        
        try{
            String destType = jmsjmx.removeJmsDestination(jndiName);

            // Send to the listner MBean a notification containing the new value of this property and 
            // the indication that the new value has to be removed from a list of values 
            ReconfiguredProp p = null;
            if (destType.equals("queue")) {
                p = new ReconfiguredProp(QUEUES, jndiName, false);
            } else if (destType.equals("topic")) {
                p = new ReconfiguredProp(TOPICS, jndiName, false);
            } else {
                return;
            }
            p.setAdd(false);
            sendReconfigNotification(++sequenceNumber, SERVICE_NAME, p);
        } catch (Exception e){
            // remove problem
            serverlog.log(BasicLevel.ERROR, "JmsService: Exception when remove destination"+e);
        }
    }

    public Set getAllJmsQueueDestinationNames() {
        HashSet result = new HashSet();
        for (Enumeration enu = jonasjms.getQueuesNames() ; enu.hasMoreElements() ;) {
            result.add(enu.nextElement());
        }
        return result;
    }

    public Set getAllJmsTopicDestinationNames(){
        HashSet result = new HashSet();
        for (Enumeration enu = jonasjms.getTopicsNames() ; enu.hasMoreElements() ;) {
            result.add(enu.nextElement());
        }
        return result;
    }

    public Set getAllJmsConnectionFactoryNames() {
        HashSet result = new HashSet();
        return result;
    }

    public Set getAllJmsQueueConnectionFactoryNames() {
        HashSet result = new HashSet();
        return result;
    }

    public Set getAllJmsTopicConnectionFactoryNames() {
        HashSet result = new HashSet();
        return result;  
    }

    public String getDefaultQueueConnectionFactoryName() {
        return jmsjmx.getDefaultQueueConnectionFactoryName(); 
    }

    public String getDefaultTopicConnectionFactoryName() {
        return jmsjmx.getDefaultTopicConnectionFactoryName(); 
    }

    public String getDefaultConnectionFactoryName() {
        return jmsjmx.getDefaultConnectionFactoryName();
    }

    public void saveConfig() {
        sendSaveNotification(++sequenceNumber, SERVICE_NAME);
    }

    // Monitoring methods

    public String getConnectionFactoryMode(String jndiName) {
        serverlog.log(BasicLevel.DEBUG, "");
        String m = null;
        try{
            m = jmsjmx.getConnectionFactoryMode(jndiName);
        } catch (IllegalStateException se) {
            // nothing to do
        } catch (Exception e){
            serverlog.log(BasicLevel.ERROR, "Exception when calling monitoring operation " + e);
        }    
        return m;
    }
 
    public Integer getPendingMessages(String jndiName) {
        serverlog.log(BasicLevel.DEBUG, "");
        Integer nb = null;
        int n;
        try{
            n = jmsjmx.getPendingMessages(jndiName);
            nb = new Integer(n);
        } catch (IllegalStateException se) {
            // nothing to do
        } catch (Exception e){
            serverlog.log(BasicLevel.ERROR, "Exception when calling monitoring operation " + e);
        }    
        return nb;
    }

    public Integer getPendingRequests(String jndiName) {
        serverlog.log(BasicLevel.DEBUG, "");
        Integer nb = null;
        int n;
        try{
            n = jmsjmx.getPendingRequests(jndiName);
            nb = new Integer(n);
        } catch (IllegalStateException se) {
            // nothing to do
        } catch (Exception e){
            serverlog.log(BasicLevel.ERROR, "Exception when calling monitoring operation " + e);
        }    
        return nb;
    }

    public Integer getSubscriptions(String jndiName) {
        serverlog.log(BasicLevel.DEBUG, "");
        Integer nb = null;
        int n;
        try{
            n = jmsjmx.getSubscriptions(jndiName);
            nb = new Integer(n);
        } catch (IllegalStateException se) {
            // nothing to do
        } catch (Exception e){
            serverlog.log(BasicLevel.ERROR, "Exception when calling monitoring operation " + e);
        }    
        return nb;
    } 

    public Boolean isMomLocal() {
        return new Boolean(isMomLocal);
    }
    public String getUrl() {
        return url;
    }

    public String getMom() {
        return momProviderClassName;
    }
}

---