JmxServiceImpl.java

package org.objectweb.jonas.jmx;

import java.io.IOException;
import java.lang.reflect.Method;
import java.net.MalformedURLException;
import java.net.URI;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.Properties;
import java.util.StringTokenizer;

import javax.management.InstanceNotFoundException;
import javax.management.JMException;
import javax.management.ListenerNotFoundException;
import javax.management.MBeanServerConnection;
import javax.management.MBeanServerNotification;
import javax.management.Notification;
import javax.management.NotificationListener;
import javax.management.ObjectName;
import javax.management.remote.JMXConnector;
import javax.management.remote.JMXConnectorFactory;
import javax.management.remote.JMXConnectorServer;
import javax.management.remote.JMXConnectorServerFactory;
import javax.management.remote.JMXServiceURL;
import javax.naming.Context;
import javax.naming.InitialContext;

import org.objectweb.carol.util.configuration.CarolConfiguration;
import org.objectweb.carol.util.configuration.CarolCurrentConfiguration;
import org.objectweb.jonas.common.Log;
import org.objectweb.jonas.discovery.DiscEvent;
import org.objectweb.jonas.service.JonasAlreadyStartedException;
import org.objectweb.jonas.service.ServiceException;
import org.objectweb.util.monolog.api.BasicLevel;
import org.objectweb.util.monolog.api.Logger;

public class JmxServiceImpl extends AbsJmxServiceImpl implements NotificationListener {

    private static final String MX4J_COMMONS_LOGGER_CLASSNAME = "mx4j.log.CommonsLogger";

    private static final String MX4J_LOG_CLASSNAME = "mx4j.log.Log";

    private static final String MX4J_LOGGER_CLASS = "mx4j.log.Logger";

    private String rmiConnectorName = null;

    public String getRmiConnectorName() {
        return this.rmiConnectorName;
    }

    private static final String JRMP = "jrmp";
    private static final String IIOP = "iiop";
    private static final String JEREMIE = "jeremie";
    private static final String CMI = "cmi";
    private JMXConnectorServer[] connectorServers = null;

    private JMXServiceURL[] connectorServerURLs = null;

    // Domain management support
    private Map managedServersToUrls = null;
    private Map managedServersToConnectors = null;
    private Map managedServersToConnections = null;

    private static Logger logger = logger = Log.getLogger("org.objectweb.jonas.jmx");
    public void doInit(Context ctx) throws ServiceException {
        // Test if MX4J CommonsLoggger class is present.
        // In this case, redirect MX4J logging to Jakarta Commons Logging.
        Class mx4jCommonsLoggerClass = null;
        try {
            mx4jCommonsLoggerClass = Thread.currentThread().getContextClassLoader().loadClass(MX4J_COMMONS_LOGGER_CLASSNAME);
            if (logger.isLoggable(BasicLevel.DEBUG)) {
                logger.log(BasicLevel.DEBUG, "Class " + MX4J_COMMONS_LOGGER_CLASSNAME + " founded");
            }
            Object o =  mx4jCommonsLoggerClass.newInstance();



            // Load Log a Logger class
            Class clazz = Thread.currentThread().getContextClassLoader().loadClass(MX4J_LOG_CLASSNAME);
            Class mx4jLoggerClass = Thread.currentThread().getContextClassLoader().loadClass(MX4J_LOGGER_CLASS);


            // Then get method redirectTo
            Method m = clazz.getMethod("redirectTo", new Class[] {mx4jLoggerClass});
            m.invoke(clazz, new Object[] {o});
            if (logger.isLoggable(BasicLevel.DEBUG)) {
                logger.log(BasicLevel.DEBUG, "MX4J logging redirected to the Jakarta commons logger");
            }
        } catch (ClassNotFoundException cnfe) {
            if (logger.isLoggable(BasicLevel.DEBUG)) {
                logger.log(BasicLevel.DEBUG, "Class " + MX4J_COMMONS_LOGGER_CLASSNAME + " not found: " + cnfe);
            }
        } catch (Exception e) {
            if (logger.isLoggable(BasicLevel.WARN)) {
                logger.log(BasicLevel.WARN, "Problem with " + MX4J_COMMONS_LOGGER_CLASSNAME + " instance creation " + e);
            }
        }
        super.doInit(ctx);
    }
    public void doStart() throws ServiceException {
        String serverName = getJonasServerName();
        try {
            // Create a RMI Connector for the JMX agent using the JOnAS RMIConnector class
            // --------------------------------------------------------------------------
            RMIConnector rmiConnector = new RMIConnectorImpl(getJmxServer());
            // Register the connector in JNDI
            InitialContext ictx = new InitialContext();
            rmiConnectorName = "RMIConnector_" + serverName;
            ictx.bind(rmiConnectorName, rmiConnector);
            ictx.close();

            // Create one or more connector servers
            // (cf. JSR 160, JMX Remote 1.0)
            // ------------------------------------
            // Determine protocols used by Carol and their configuration
            String s = CarolConfiguration.getProtocols();
            StringTokenizer st = new StringTokenizer(s, ",", false);
            HashMap myCarolConfig = new HashMap();
            while (st.hasMoreElements()) {
                CarolCurrentConfiguration carolConfig = CarolCurrentConfiguration.getCurrent();
                String protocol = st.nextToken().trim();
                //if (!protocol.equals(CMI)) {
                    myCarolConfig.put(protocol, carolConfig.getRMIProperties(protocol));
                //}
            }
            int nbProtocols = myCarolConfig.size();
            // Create a JMXServiceURL and a JMXConnectorServer per protocol
            connectorServerURLs = new JMXServiceURL[nbProtocols];
            connectorServers = new JMXConnectorServer[nbProtocols];
            // index over connectorServerURLs
            int i = 0;
            String serviceURL = null;
            // arguments needed to construct a JMXServiceURL
            String myProtocol = null;
            String myHost = null;
            String myPort = null;
            String myName = null;
            for (Iterator it = myCarolConfig.keySet().iterator(); it.hasNext();) {
                String carolProtocol = (String) it.next();
                Properties props = (Properties) myCarolConfig.get(carolProtocol);
                String sCarolURL = props.getProperty(Context.PROVIDER_URL);
                URI carolURL = new URI(sCarolURL);
                myHost = carolURL.getHost();
                myPort = String.valueOf(carolURL.getPort());
                if (carolProtocol.equals(JRMP)) {
                    // Treat JRMP case
                    myProtocol = "jrmp";
                    myName = "jrmpconnector_" + serverName;
                    serviceURL = "service:jmx:rmi://" + myHost + "/jndi/rmi://"
                        + myHost + ":" + myPort + "/"
                        + myName;
                } else if (carolProtocol.equals(JEREMIE)) {
                    // Treat JEREMIE case
                    myProtocol = "jeremie";
                    myName = "jeremieconnector_" + serverName;
                    // PATCH for JEREMIE ONLY !!!!
                    //int anyNumber = 5;
                    //thePort = thePort + anyNumber;
                    serviceURL = "service:jmx:rmi://" + myHost + "/jndi/jrmi://"
                        + myHost + ":" + myPort + "/"
                        + myName;
                } else if (carolProtocol.equals(IIOP)) {
                    // Treat IIOP case
                    myProtocol = "iiop";
                    myName = "iiopconnector_" + serverName;
                    serviceURL = "service:jmx:iiop://" + myHost + "/jndi/" + myName;
                    props.put("java.naming.corba.orb", new InitialContext().lookup("java:comp/ORB"));

                } else if (carolProtocol.equals(CMI)) {
                    // Treat JRMP case
                    myProtocol = "cmi";
                    myName = "cmiconnector_" + serverName;
                    serviceURL = "service:jmx:rmi://" + myHost + "/jndi/cmi://"
                        + myHost + ":" + myPort + "/"
                        + myName;
                }
                JMXServiceURL url = new JMXServiceURL(serviceURL);
                // Cast to Map is required for JDK 1.5
                JMXConnectorServer connectorServer = JMXConnectorServerFactory.newJMXConnectorServer(url, (Map) props, null);
                connectorServers[i] = connectorServer;
                // Create the MBean associated to the connector
                String connectorObjectName = "connector_" + carolProtocol;
                ObjectName connectorServerName = JonasObjectName.jmxConnectorServer(myProtocol, connectorObjectName);
                getJmxServer().registerMBean(connectorServer, connectorServerName);
                // Start the JMXConnectorServer
                try {
                    connectorServer.start();
                    connectorServerURLs[i] = connectorServer.getAddress();
                } catch (IllegalArgumentException e) {
                    throw e;
                }
                i++;
            }
        } catch (javax.naming.NameAlreadyBoundException ne) {
            getLogger().log(BasicLevel.DEBUG, "Cannot start JMX service " + ne);
            throw new JonasAlreadyStartedException();
        } catch (Exception e) {
            getLogger().log(BasicLevel.DEBUG, "Cannot start JMX service " + e);
            throw new ServiceException("Cannot start JMX service", e);
        }

        // Registers as listener to notifications emitted by MBeanServerDelegate
        // MBeanServerDelegate is the MBean which sends registration/unregistration notifications
        try {
            ObjectName delegate = ObjectName.getInstance("JMImplementation:type=MBeanServerDelegate");
            getJmxServer().addNotificationListener(delegate, this, null, null);
        } catch (JMException me) {
            if (logger.isLoggable(BasicLevel.DEBUG)) {
                logger.log(BasicLevel.DEBUG, "JMX service could not be added as notification listener for MBeanServerNotifications " +
                        "related to the REGISTRATION or UNREGISTRETION of JOnAS management MBeans");
            }
        }
        // Create data structures allowing to manage the servers in the domain
        managedServersToConnectors = new HashMap();
        managedServersToConnections = new HashMap();
        managedServersToUrls = new HashMap();
    }
    public void doStop() {
        try {
            InitialContext ictx = new InitialContext();
            ictx.unbind("RMIConnector_" + getJonasServerName());
            ictx.close();

            // Desactivates the connector server, that is, stops listening for client connections.
            // Calling this method will also close all client connections that were made by this server.
            // Being a potentialy slow operation, we keep it commented for the moment
            for (int i = 0; i < connectorServers.length; i++) {
                connectorServers[i].stop();
            }
        } catch (Exception e) {
            getLogger().log(BasicLevel.ERROR, "Cannot Unbind Jmx RMI Connector" + e);
        }
        // Unregister some MBeans
        ObjectName domainOn = J2eeObjectName.J2EEDomain(getDomainName());
        try {
            getJmxServer().unregisterMBean(domainOn);
        } catch (Exception e) {
            getLogger().log(BasicLevel.INFO, "Cannot unregister JEEDomain MBean:" + domainOn.toString());
        }
        ObjectName serverOn = J2eeObjectName.J2EEServer(getDomainName(), getJonasServerName());
        try {
            getJmxServer().unregisterMBean(serverOn);
        } catch (Exception e) {
            getLogger().log(BasicLevel.INFO, "Cannot unregister JEEServer MBean:" + serverOn.toString());
        }
        // Remove internal references to the MBeanServer.
        releaseJmxServer();

        getLogger().log(BasicLevel.DEBUG, "JMX Service stopped");
    }


    public JMXServiceURL[] getConnectorServerURLs() {
        return this.connectorServerURLs;
    }

    public void handleNotification(Notification notification, Object handback) {
        if (notification instanceof MBeanServerNotification) {
            String type = notification.getType();
            // ObjectName of the MBean that caused the notification
            ObjectName registeredOn = ((MBeanServerNotification) notification).getMBeanName();
            String name = registeredOn.getKeyProperty("name");
            // The names below are defined in JonasObjectName class
            if ((name != null) && (name.equals("discoveryEnroller") || name.equals("discoveryClient"))) {
                if (type.equals(MBeanServerNotification.REGISTRATION_NOTIFICATION)) {
                    try {
                        // register myself as listener of notifications emitted by the Enroller or DiscoveryClient MBean
                        getJmxServer().addNotificationListener(registeredOn, this, null, null);
                        if (logger.isLoggable(BasicLevel.DEBUG)) {
                            logger.log(BasicLevel.DEBUG, "J2EEDomain (this) registered as listener to notifs emitted by " + registeredOn);
                        }
                    } catch (InstanceNotFoundException e) {
                        // TODO Auto-generated catch block
                        e.printStackTrace();
                    }
                }
                if (type.equals(MBeanServerNotification.UNREGISTRATION_NOTIFICATION)) {
                    try {
                        // unregister myself as listener of notifications emitted by the Enroller or DiscoveryClient MBean
                        getJmxServer().removeNotificationListener(registeredOn, this);
                        if (logger.isLoggable(BasicLevel.DEBUG)) {
                            logger.log(BasicLevel.DEBUG, "J2EEDomain (this) removed listener for notifs emitted by " + registeredOn);
                        }
                    } catch (InstanceNotFoundException e) {
                        // TODO Auto-generated catch block
                        e.printStackTrace();
                    } catch (ListenerNotFoundException e) {
                        // TODO Auto-generated catch block
                        e.printStackTrace();
                    }
                }
            }
        } else {
            // Tread discovery notification
            String type = notification.getType();
            String message = notification.getMessage();
            DiscEvent userData = (DiscEvent) notification.getUserData();
            String source = ((ObjectName) notification.getSource()).toString();
            String state = userData.getState();
            if (logger.isLoggable(BasicLevel.DEBUG)) {
                logger.log(BasicLevel.DEBUG, "Treat notification:");
                logger.log(BasicLevel.DEBUG, "- source: " + source);
                logger.log(BasicLevel.DEBUG, "- type: " + type);
                logger.log(BasicLevel.DEBUG, "- data: ");
                logger.log(BasicLevel.DEBUG, "--- state: " + state);
                logger.log(BasicLevel.DEBUG, "--- serverName : " + userData.getServerName());
                logger.log(BasicLevel.DEBUG, "--- domainName : " + userData.getDomainName());
                if (userData.getConnectorURL() != null) {
                    String[] urls = userData.getConnectorURL();
                    for (int i = 0; i < urls.length; i++) {
                        logger.log(BasicLevel.DEBUG, "--- urls : " + urls[i]);
                    }
                }
                logger.log(BasicLevel.DEBUG, "");
            }
            if (state.equals(DiscEvent.RUNNING)) {
                String[] urls = userData.getConnectorURL();
                for (int i = 0; i < urls.length; i++) {
                    addServer(userData.getDomainName(), userData.getServerName(), urls[i]);
                }
            } else if (state.equals(DiscEvent.STOPPING)) {
                removeServer(userData.getDomainName(), userData.getServerName());
            }
        }
    }
    public void addServer(String domainName, String serverName, String connectorServerURL) {
        if (logger.isLoggable(BasicLevel.DEBUG)) {
            logger.log(BasicLevel.DEBUG, "domain name: " + domainName + ", server name: " + serverName + ", url: " + connectorServerURL);
        }
        if (domainName.equals(getDomainName())) {
            // Server thet belongs to the current management domain
            if (managedServersToUrls.containsKey(serverName)) {
                // This server is already known
                ArrayList urls = (ArrayList) managedServersToUrls.get(serverName);
                if (!urls.contains(connectorServerURL)) {
                    // Add this new connectorServerURL in the connectors list
                    urls.add(connectorServerURL);
                }
            } else {
                // Add new server
                ArrayList urls = new ArrayList();
                urls.add(connectorServerURL);
                managedServersToUrls.put(serverName, urls);
            }
            // try to create a connection
            boolean created = createConnection(serverName, connectorServerURL);
        } else {
            if (logger.isLoggable(BasicLevel.WARN)) {
                logger.log(BasicLevel.WARN, "The server named " + serverName + " was not started in the management domain "
                        + getDomainName() + ", but in management domain " + domainName);
            }
        }
    }

    private boolean createConnection(String serverName, String connectorServerURL) {
        boolean created = false;
        MBeanServerConnection connection;
        // create a connector client for the connector server at the given url
        JMXConnector connector = null;
        try {
            JMXServiceURL url = new JMXServiceURL(connectorServerURL);
            connector = JMXConnectorFactory.newJMXConnector(url, null);
            connector.connect(null);
            connection = connector.getMBeanServerConnection();
        } catch (MalformedURLException e) {
            // there is no provider for the protocol in url
            connection = null;
        } catch (IOException e) {
            // connector client or connection cannot be made because of a communication problem.
            connection = null;
        } catch (java.lang.SecurityException e) {
            // connection cannot be made for security reasons
            connection = null;
        }
        if (connection != null) {
            // OK, connection was established to the connector server.
            testConnection(connection, serverName);
            if (!managedServersToConnections.containsKey(serverName)) {
                // add new connection and connector to managed data structure
                managedServersToConnections.put(serverName, connection);
                managedServersToConnectors.put(serverName, connector);
                created = true;
                logger.log(BasicLevel.INFO, "First MBeanServerConnection created for connecting to server " + serverName + " using URL: " + connectorServerURL);
            } else {
                // A connection already exists.
                // We have the following possibilities:
                // - ignore this connection (close it), so continue to use the old one
                // - replace the old one by the new one (close the old)
                // - keep several connections
                //
                // Currently we keep only one connection, the new one (second approach)
                JMXConnector oldConnector = (JMXConnector) managedServersToConnectors.get(serverName);
                try {
                    oldConnector.close();
                } catch (IOException e1) {
                }
                managedServersToConnectors.put(serverName, connector);
                managedServersToConnections.put(serverName, connection);
                created = true;
                logger.log(BasicLevel.INFO, "New MBeanServerConnection created for connecting to server " + serverName + " using URL: " + connectorServerURL + "(replaces the old connection)");
            }
        } else {
            if (connector != null) {
                try {
                    connector.close();
                } catch (IOException e1) {
                }
            }
        }
        return created;
    }

    private boolean testConnection(MBeanServerConnection connection, String serverName) {
        boolean ok = false;
        try {
            Integer nb = connection.getMBeanCount();
            /*
            System.out.println("IN TEST connection: got " + nb.toString() + " MBeans in the JMX server of JOnAS server " + serverName);
            */
            ok = true;
        } catch (IOException ioe) {
            ioe.printStackTrace();
        }
        return ok;
    }

    public void removeServer(String domainName, String serverName) {
        if (domainName.equals(getDomainName())) {
            // Server thet belongs to the current management domain
            managedServersToConnections.remove(serverName);
            JMXConnector connector = (JMXConnector) managedServersToConnectors.remove(serverName);
            if (connector != null) {
            try {
                connector.close();
            } catch (IOException e1) {
            }
            }
            managedServersToUrls.remove(serverName);
        } else {
            if (logger.isLoggable(BasicLevel.WARN)) {
                logger.log(BasicLevel.WARN, "The server named " + serverName + " was not started in the management domain "
                        + getDomainName() + ", but in management domain " + domainName);
            }
        }
    }
    public MBeanServerConnection getServerConnection(String serverName) {
        return (MBeanServerConnection) managedServersToConnections.get(serverName);
    }
}

---