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.plugin;
018    
019    import org.apache.activemq.advisory.AdvisorySupport;
020    import org.apache.activemq.broker.Broker;
021    import org.apache.activemq.broker.BrokerFilter;
022    import org.apache.activemq.broker.BrokerService;
023    import org.apache.activemq.broker.ConnectionContext;
024    import org.apache.activemq.broker.ProducerBrokerExchange;
025    import org.apache.activemq.broker.region.Destination;
026    import org.apache.activemq.broker.region.DestinationStatistics;
027    import org.apache.activemq.broker.region.RegionBroker;
028    import org.apache.activemq.command.ActiveMQDestination;
029    import org.apache.activemq.command.ActiveMQMapMessage;
030    import org.apache.activemq.command.Message;
031    import org.apache.activemq.command.MessageId;
032    import org.apache.activemq.command.ProducerId;
033    import org.apache.activemq.command.ProducerInfo;
034    import org.apache.activemq.state.ProducerState;
035    import org.apache.activemq.usage.SystemUsage;
036    import org.apache.activemq.util.IdGenerator;
037    import org.apache.activemq.util.LongSequenceGenerator;
038    import org.apache.commons.logging.Log;
039    import org.apache.commons.logging.LogFactory;
040    import java.io.File;
041    import java.net.URI;
042    import java.util.Set;
043    /**
044     * A StatisticsBroker You can retrieve a Map Message for a Destination - or
045     * Broker containing statistics as key-value pairs The message must contain a
046     * replyTo Destination - else its ignored
047     * 
048     */
049    public class StatisticsBroker extends BrokerFilter {
050        private static Log LOG = LogFactory.getLog(StatisticsBroker.class);
051        static final String STATS_DESTINATION_PREFIX = "ActiveMQ.Statistics.Destination";
052        static final String STATS_BROKER_PREFIX = "ActiveMQ.Statistics.Broker";
053        private static final IdGenerator ID_GENERATOR = new IdGenerator();
054        private final LongSequenceGenerator messageIdGenerator = new LongSequenceGenerator();
055        protected final ProducerId advisoryProducerId = new ProducerId();
056    
057        /**
058         * 
059         * Constructor
060         * 
061         * @param next
062         */
063        public StatisticsBroker(Broker next) {
064            super(next);
065            this.advisoryProducerId.setConnectionId(ID_GENERATOR.generateId());
066        }
067    
068        /**
069         * Sets the persistence mode
070         * 
071         * @see org.apache.activemq.broker.BrokerFilter#send(org.apache.activemq.broker.ProducerBrokerExchange,
072         *      org.apache.activemq.command.Message)
073         */
074        public void send(ProducerBrokerExchange producerExchange, Message messageSend) throws Exception {
075            ActiveMQDestination msgDest = messageSend.getDestination();
076            ActiveMQDestination replyTo = messageSend.getReplyTo();
077            if (replyTo != null) {
078                String physicalName = msgDest.getPhysicalName();
079                boolean destStats = physicalName.regionMatches(true, 0, STATS_DESTINATION_PREFIX, 0,
080                        STATS_DESTINATION_PREFIX.length());
081                boolean brokerStats = physicalName.regionMatches(true, 0, STATS_BROKER_PREFIX, 0, STATS_BROKER_PREFIX
082                        .length());
083                if (destStats) {
084                    String queueryName = physicalName.substring(STATS_DESTINATION_PREFIX.length(), physicalName.length());
085                    ActiveMQDestination queryDest = ActiveMQDestination.createDestination(queueryName,msgDest.getDestinationType());
086                    Set<Destination> set = getDestinations(queryDest);
087                    for (Destination dest : set) {
088                        DestinationStatistics stats = dest.getDestinationStatistics();
089                        if (stats != null) {
090                            ActiveMQMapMessage statsMessage = new ActiveMQMapMessage();
091                            statsMessage.setString("destinationName", dest.getActiveMQDestination().toString());
092                            statsMessage.setLong("size", stats.getMessages().getCount());
093                            statsMessage.setLong("enqueueCount", stats.getEnqueues().getCount());
094                            statsMessage.setLong("dequeueCount", stats.getDequeues().getCount());
095                            statsMessage.setLong("dispatchCount", stats.getDispatched().getCount());
096                            statsMessage.setLong("expiredCount", stats.getExpired().getCount());
097                            statsMessage.setLong("inflightCount", stats.getInflight().getCount());
098                            statsMessage.setLong("messagesCached", stats.getMessagesCached().getCount());
099                            statsMessage.setInt("memoryPercentUsage", dest.getMemoryUsage().getPercentUsage());
100                            statsMessage.setLong("memoryUsage", dest.getMemoryUsage().getUsage());
101                            statsMessage.setLong("memoryLimit", dest.getMemoryUsage().getLimit());
102                            statsMessage.setDouble("averageEnqueueTime", stats.getProcessTime().getAverageTime());
103                            statsMessage.setDouble("maxEnqueueTime", stats.getProcessTime().getMaxTime());
104                            statsMessage.setDouble("minEnqueueTime", stats.getProcessTime().getMinTime());
105                            statsMessage.setLong("consumerCount", stats.getConsumers().getCount());
106                            statsMessage.setLong("producerCount", stats.getProducers().getCount());
107                            sendStats(producerExchange.getConnectionContext(), statsMessage, replyTo);
108                        }
109                    }
110                } else if (brokerStats) {
111                    ActiveMQMapMessage statsMessage = new ActiveMQMapMessage();
112                    BrokerService brokerService = getBrokerService();
113                    RegionBroker regionBroker = (RegionBroker) brokerService.getRegionBroker();
114                    SystemUsage systemUsage = brokerService.getSystemUsage();
115                    DestinationStatistics stats = regionBroker.getDestinationStatistics();
116                    statsMessage.setString("brokerName", regionBroker.getBrokerName());
117                    statsMessage.setString("brokerId", regionBroker.getBrokerId().toString());
118                    statsMessage.setLong("size", stats.getMessages().getCount());
119                    statsMessage.setLong("enqueueCount", stats.getEnqueues().getCount());
120                    statsMessage.setLong("dequeueCount", stats.getDequeues().getCount());
121                    statsMessage.setLong("dispatchCount", stats.getDispatched().getCount());
122                    statsMessage.setLong("expiredCount", stats.getExpired().getCount());
123                    statsMessage.setLong("inflightCount", stats.getInflight().getCount());
124                    statsMessage.setLong("messagesCached", stats.getMessagesCached().getCount());
125                    statsMessage.setInt("memoryPercentUsage", systemUsage.getMemoryUsage().getPercentUsage());
126                    statsMessage.setLong("memoryUsage", systemUsage.getMemoryUsage().getUsage());
127                    statsMessage.setLong("memoryLimit", systemUsage.getMemoryUsage().getLimit());
128                    statsMessage.setInt("storePercentUsage", systemUsage.getStoreUsage().getPercentUsage());
129                    statsMessage.setLong("storeUsage", systemUsage.getStoreUsage().getUsage());
130                    statsMessage.setLong("storeLimit", systemUsage.getStoreUsage().getLimit());
131                    statsMessage.setInt("tempPercentUsage", systemUsage.getTempUsage().getPercentUsage());
132                    statsMessage.setLong("tempUsage", systemUsage.getTempUsage().getUsage());
133                    statsMessage.setLong("tempLimit", systemUsage.getTempUsage().getLimit());
134                    statsMessage.setDouble("averageEnqueueTime", stats.getProcessTime().getAverageTime());
135                    statsMessage.setDouble("maxEnqueueTime", stats.getProcessTime().getMaxTime());
136                    statsMessage.setDouble("minEnqueueTime", stats.getProcessTime().getMinTime());
137                    statsMessage.setLong("consumerCount", stats.getConsumers().getCount());
138                    statsMessage.setLong("producerCount", stats.getProducers().getCount());
139                    String answer = brokerService.getTransportConnectorURIsAsMap().get("tcp");
140                    answer = answer != null ? answer : "";
141                    statsMessage.setString("openwire", answer);
142                    answer = brokerService.getTransportConnectorURIsAsMap().get("stomp");
143                    answer = answer != null ? answer : "";
144                    statsMessage.setString("stomp", answer);
145                    answer = brokerService.getTransportConnectorURIsAsMap().get("ssl");
146                    answer = answer != null ? answer : "";
147                    statsMessage.setString("ssl", answer);
148                    answer = brokerService.getTransportConnectorURIsAsMap().get("stomp+ssl");
149                    answer = answer != null ? answer : "";
150                    statsMessage.setString("stomp+ssl", answer);
151                    URI uri = brokerService.getVmConnectorURI();
152                    answer = uri != null ? uri.toString() : "";
153                    statsMessage.setString("vm", answer);
154                    File file = brokerService.getDataDirectoryFile();
155                    answer = file != null ? file.getCanonicalPath() : "";
156                    statsMessage.setString("dataDirectory", answer);
157                    sendStats(producerExchange.getConnectionContext(), statsMessage, replyTo);
158                } else {
159                    super.send(producerExchange, messageSend);
160                }
161            } else {
162                super.send(producerExchange, messageSend);
163            }
164        }
165    
166        public void start() throws Exception {
167            super.start();
168            LOG.info("Starting StatisticsBroker");
169        }
170    
171        public void stop() throws Exception {
172            super.stop();
173        }
174    
175        protected void sendStats(ConnectionContext context, ActiveMQMapMessage msg, ActiveMQDestination replyTo)
176                throws Exception {
177            msg.setPersistent(false);
178            msg.setType(AdvisorySupport.ADIVSORY_MESSAGE_TYPE);
179            msg.setMessageId(new MessageId(this.advisoryProducerId, this.messageIdGenerator.getNextSequenceId()));
180            msg.setDestination(replyTo);
181            msg.setResponseRequired(false);
182            msg.setProducerId(this.advisoryProducerId);
183            boolean originalFlowControl = context.isProducerFlowControl();
184            final ProducerBrokerExchange producerExchange = new ProducerBrokerExchange();
185            producerExchange.setConnectionContext(context);
186            producerExchange.setMutable(true);
187            producerExchange.setProducerState(new ProducerState(new ProducerInfo()));
188            try {
189                context.setProducerFlowControl(false);
190                this.next.send(producerExchange, msg);
191            } finally {
192                context.setProducerFlowControl(originalFlowControl);
193            }
194        }
195    }