001    /**
002     * Licensed to the Apache Software Foundation (ASF) under one or more
003     * contributor license agreements.  See the NOTICE file distributed with
004     * this work for additional information regarding copyright ownership.
005     * The ASF licenses this file to You under the Apache License, Version 2.0
006     * (the "License"); you may not use this file except in compliance with
007     * the License.  You may obtain a copy of the License at
008     *
009     *      http://www.apache.org/licenses/LICENSE-2.0
010     *
011     * Unless required by applicable law or agreed to in writing, software
012     * distributed under the License is distributed on an "AS IS" BASIS,
013     * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
014     * See the License for the specific language governing permissions and
015     * limitations under the License.
016     */
017    package org.apache.activemq.web;
018    
019    import java.io.IOException;
020    import java.util.ArrayList;
021    import java.util.Collection;
022    import java.util.HashMap;
023    import java.util.List;
024    import java.util.Map;
025    import java.util.Set;
026    
027    import javax.management.MBeanServerConnection;
028    import javax.management.MBeanServerInvocationHandler;
029    import javax.management.MalformedObjectNameException;
030    import javax.management.ObjectName;
031    import javax.management.remote.JMXConnector;
032    import javax.management.remote.JMXConnectorFactory;
033    import javax.management.remote.JMXServiceURL;
034    
035    import org.apache.activemq.broker.jmx.BrokerViewMBean;
036    import org.apache.activemq.broker.jmx.ManagementContext;
037    import org.apache.activemq.broker.jmx.QueueViewMBean;
038    import org.apache.activemq.command.ActiveMQDestination;
039    import org.apache.activemq.web.config.WebConsoleConfiguration;
040    import org.apache.commons.logging.Log;
041    import org.apache.commons.logging.LogFactory;
042    
043    /**
044     * A {@link BrokerFacade} which uses a JMX-Connection to communicate with a
045     * broker
046     * 
047     * @version $Revision: 1.1 $
048     */
049    public class RemoteJMXBrokerFacade extends BrokerFacadeSupport {
050        
051        private static final transient Log LOG = LogFactory.getLog(RemoteJMXBrokerFacade.class);
052        
053        private String brokerName;
054        private JMXConnector connector;
055        private WebConsoleConfiguration configuration;
056    
057        public void setBrokerName(String brokerName) {
058            this.brokerName = brokerName;
059        }
060    
061        public WebConsoleConfiguration getConfiguration() {
062                    return configuration;
063            }
064    
065            public void setConfiguration(WebConsoleConfiguration configuration) {
066                    this.configuration = configuration;
067            }
068    
069            /**
070         * Shutdown this facade aka close any open connection.
071         */
072        public void shutdown() {
073            closeConnection();
074        }
075        
076        private ObjectName getBrokerObjectName(MBeanServerConnection connection)
077                            throws IOException, MalformedObjectNameException {
078                    Set<ObjectName> brokers = findBrokers(connection);
079                    if (brokers.size() == 0) {
080                            throw new IOException("No broker could be found in the JMX.");
081                    }
082                    ObjectName name = brokers.iterator().next();
083                    return name;
084            }
085    
086        public BrokerViewMBean getBrokerAdmin() throws Exception {
087            MBeanServerConnection connection = getMBeanServerConnection();
088    
089            Set brokers = findBrokers(connection);
090            if (brokers.size() == 0) {
091                throw new IOException("No broker could be found in the JMX.");
092            }
093            ObjectName name = (ObjectName)brokers.iterator().next();
094            BrokerViewMBean mbean = (BrokerViewMBean)MBeanServerInvocationHandler.newProxyInstance(connection, name, BrokerViewMBean.class, true);
095            return mbean;
096        }
097    
098        public String getBrokerName() throws Exception,
099                            MalformedObjectNameException {
100            return getBrokerAdmin().getBrokerName();
101        }
102        
103        protected MBeanServerConnection getMBeanServerConnection() throws Exception {
104            JMXConnector connector = this.connector;
105            if (isConnectionActive(connector)) {
106                return connector.getMBeanServerConnection();
107            }
108    
109            synchronized (this) {
110                closeConnection();
111    
112                LOG.debug("Creating a new JMX-Connection to the broker");
113                this.connector = createConnection();
114                return this.connector.getMBeanServerConnection();
115            }
116        }
117    
118        protected boolean isConnectionActive(JMXConnector connector) {
119            if (connector == null) {
120                return false;
121            }
122    
123            try {
124                MBeanServerConnection connection = connector.getMBeanServerConnection();
125                int brokerCount = findBrokers(connection).size();
126                return brokerCount > 0;
127            } catch (Exception e) {
128                return false;
129            }
130        }
131    
132        protected JMXConnector createConnection() {
133    
134            Map<String, Object> env = new HashMap<String, Object>();
135                    if (this.configuration.getJmxUser() != null) {
136                            env.put("jmx.remote.credentials", new String[] {
137                                            this.configuration.getJmxUser(),
138                                            this.configuration.getJmxPassword() });
139                    }
140            Collection<JMXServiceURL> jmxUrls = this.configuration.getJmxUrls();
141    
142            Exception exception = null;
143                    for (JMXServiceURL url : jmxUrls) {
144                            try {
145                                    JMXConnector connector = JMXConnectorFactory.connect(url, env);
146                                    connector.connect();
147                                    MBeanServerConnection connection = connector
148                                                    .getMBeanServerConnection();
149    
150                                    Set<ObjectName> brokers = findBrokers(connection);
151                                    if (brokers.size() > 0) {
152                                            LOG.info("Connected via JMX to the broker at " + url);
153                                            return connector;
154                                    }
155                            } catch (Exception e) {
156                                    // Keep the exception for later
157                                    exception = e;
158                            }
159                    }
160                    if (exception != null) {
161                            if (exception instanceof RuntimeException) {
162                                    throw (RuntimeException) exception;
163                            } else {
164                                    throw new RuntimeException(exception);
165                            }
166                    }
167                    throw new IllegalStateException("No broker is found at any of the "
168                                    + jmxUrls.size() + " configured urls");
169            }
170    
171        protected synchronized void closeConnection() {
172            if (connector != null) {
173                try {
174                    LOG.debug("Closing a connection to a broker (" + connector.getConnectionId() + ")");
175    
176                    connector.close();
177                } catch (IOException e) {
178                    // Ignore the exception, since it most likly won't matter
179                    // anymore
180                }
181            }
182        }
183    
184            /**
185             * Finds all ActiveMQ-Brokers registered on a certain JMX-Server or, if a
186             * JMX-BrokerName has been set, the broker with that name.
187             * 
188             * @param connection
189             *            not <code>null</code>
190             * @return Set with ObjectName-elements
191             * @throws IOException
192             * @throws MalformedObjectNameException
193             */
194            @SuppressWarnings("unchecked")
195            protected Set<ObjectName> findBrokers(MBeanServerConnection connection)
196                            throws IOException, MalformedObjectNameException {
197                    ObjectName name;
198                    if (this.brokerName == null) {
199                            name = new ObjectName("org.apache.activemq:Type=Broker,*");
200                    } else {
201                            name = new ObjectName("org.apache.activemq:BrokerName="
202                                            + this.brokerName + ",Type=Broker");
203                    }
204    
205                    Set<ObjectName> brokers = connection.queryNames(name, null);
206                    return brokers;
207            }
208            
209            public void purgeQueue(ActiveMQDestination destination) throws Exception {
210                    QueueViewMBean queue = getQueue(destination.getPhysicalName());
211                    queue.purge();
212            }
213            
214            public ManagementContext getManagementContext() {
215                    throw new IllegalStateException("not supported");
216            }
217    
218            
219            @SuppressWarnings("unchecked")
220            protected <T> Collection<T> getManagedObjects(ObjectName[] names,
221                            Class<T> type) {
222                    MBeanServerConnection connection;
223                    try {
224                            connection = getMBeanServerConnection();
225                    } catch (Exception e) {
226                            throw new RuntimeException(e);
227                    }
228    
229                    List<T> answer = new ArrayList<T>();
230                    if (connection != null) {
231                            for (int i = 0; i < names.length; i++) {
232                                    ObjectName name = names[i];
233                                    T value = (T) MBeanServerInvocationHandler.newProxyInstance(
234                                                    connection, name, type, true);
235                                    if (value != null) {
236                                            answer.add(value);
237                                    }
238                            }
239                    }
240                    return answer;
241        }
242    
243    }