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 */
017package org.apache.activemq.broker.jmx;
018
019import java.io.IOException;
020import java.net.URISyntaxException;
021import java.util.ArrayList;
022import java.util.Collections;
023import java.util.HashMap;
024import java.util.Iterator;
025import java.util.List;
026import java.util.Map;
027
028import javax.jms.Connection;
029import javax.jms.InvalidSelectorException;
030import javax.jms.MessageProducer;
031import javax.jms.Session;
032import javax.management.MalformedObjectNameException;
033import javax.management.ObjectName;
034import javax.management.openmbean.CompositeData;
035import javax.management.openmbean.CompositeDataSupport;
036import javax.management.openmbean.CompositeType;
037import javax.management.openmbean.OpenDataException;
038import javax.management.openmbean.TabularData;
039import javax.management.openmbean.TabularDataSupport;
040import javax.management.openmbean.TabularType;
041
042import org.apache.activemq.ActiveMQConnectionFactory;
043import org.apache.activemq.broker.jmx.OpenTypeSupport.OpenTypeFactory;
044import org.apache.activemq.broker.region.Destination;
045import org.apache.activemq.broker.region.Subscription;
046import org.apache.activemq.broker.region.policy.AbortSlowConsumerStrategy;
047import org.apache.activemq.broker.region.policy.SlowConsumerStrategy;
048import org.apache.activemq.command.ActiveMQDestination;
049import org.apache.activemq.command.ActiveMQMessage;
050import org.apache.activemq.command.ActiveMQTextMessage;
051import org.apache.activemq.command.Message;
052import org.apache.activemq.filter.BooleanExpression;
053import org.apache.activemq.filter.MessageEvaluationContext;
054import org.apache.activemq.selector.SelectorParser;
055import org.apache.activemq.store.MessageStore;
056import org.apache.activemq.util.URISupport;
057import org.slf4j.Logger;
058import org.slf4j.LoggerFactory;
059
060public class DestinationView implements DestinationViewMBean {
061    private static final Logger LOG = LoggerFactory.getLogger(DestinationViewMBean.class);
062    protected final Destination destination;
063    protected final ManagedRegionBroker broker;
064
065    public DestinationView(ManagedRegionBroker broker, Destination destination) {
066        this.broker = broker;
067        this.destination = destination;
068    }
069
070    public void gc() {
071        destination.gc();
072    }
073
074    @Override
075    public String getName() {
076        return destination.getName();
077    }
078
079    @Override
080    public void resetStatistics() {
081        destination.getDestinationStatistics().reset();
082    }
083
084    @Override
085    public long getEnqueueCount() {
086        return destination.getDestinationStatistics().getEnqueues().getCount();
087    }
088
089    @Override
090    public long getDequeueCount() {
091        return destination.getDestinationStatistics().getDequeues().getCount();
092    }
093
094    @Override
095    public long getForwardCount() {
096        return destination.getDestinationStatistics().getForwards().getCount();
097    }
098
099    @Override
100    public long getDispatchCount() {
101        return destination.getDestinationStatistics().getDispatched().getCount();
102    }
103
104    @Override
105    public long getInFlightCount() {
106        return destination.getDestinationStatistics().getInflight().getCount();
107    }
108
109    @Override
110    public long getExpiredCount() {
111        return destination.getDestinationStatistics().getExpired().getCount();
112    }
113
114    @Override
115    public long getConsumerCount() {
116        return destination.getDestinationStatistics().getConsumers().getCount();
117    }
118
119    @Override
120    public long getQueueSize() {
121        return destination.getDestinationStatistics().getMessages().getCount();
122    }
123
124    @Override
125    public long getStoreMessageSize() {
126        MessageStore messageStore = destination.getMessageStore();
127        return messageStore != null ? messageStore.getMessageStoreStatistics().getMessageSize().getTotalSize() : 0;
128    }
129
130    public long getMessagesCached() {
131        return destination.getDestinationStatistics().getMessagesCached().getCount();
132    }
133
134    @Override
135    public int getMemoryPercentUsage() {
136        return destination.getMemoryUsage().getPercentUsage();
137    }
138
139    @Override
140    public long getMemoryUsageByteCount() {
141        return destination.getMemoryUsage().getUsage();
142    }
143
144    @Override
145    public long getMemoryLimit() {
146        return destination.getMemoryUsage().getLimit();
147    }
148
149    @Override
150    public void setMemoryLimit(long limit) {
151        destination.getMemoryUsage().setLimit(limit);
152    }
153
154    @Override
155    public double getAverageEnqueueTime() {
156        return destination.getDestinationStatistics().getProcessTime().getAverageTime();
157    }
158
159    @Override
160    public long getMaxEnqueueTime() {
161        return destination.getDestinationStatistics().getProcessTime().getMaxTime();
162    }
163
164    @Override
165    public long getMinEnqueueTime() {
166        return destination.getDestinationStatistics().getProcessTime().getMinTime();
167    }
168
169    /**
170     * @return the average size of a message (bytes)
171     */
172    @Override
173    public long getAverageMessageSize() {
174        // we are okay with the size without decimals so cast to long
175        return (long) destination.getDestinationStatistics().getMessageSize().getAverageSize();
176    }
177
178    /**
179     * @return the max size of a message (bytes)
180     */
181    @Override
182    public long getMaxMessageSize() {
183        return destination.getDestinationStatistics().getMessageSize().getMaxSize();
184    }
185
186    /**
187     * @return the min size of a message (bytes)
188     */
189    @Override
190    public long getMinMessageSize() {
191        return destination.getDestinationStatistics().getMessageSize().getMinSize();
192    }
193
194
195    @Override
196    public boolean isPrioritizedMessages() {
197        return destination.isPrioritizedMessages();
198    }
199
200    @Override
201    public CompositeData[] browse() throws OpenDataException {
202        try {
203            return browse(null);
204        } catch (InvalidSelectorException e) {
205            // should not happen.
206            throw new RuntimeException(e);
207        }
208    }
209
210    @Override
211    public CompositeData[] browse(String selector) throws OpenDataException, InvalidSelectorException {
212        Message[] messages = destination.browse();
213        ArrayList<CompositeData> c = new ArrayList<CompositeData>();
214
215        MessageEvaluationContext ctx = new MessageEvaluationContext();
216        ctx.setDestination(destination.getActiveMQDestination());
217        BooleanExpression selectorExpression = selector == null ? null : SelectorParser.parse(selector);
218
219        for (int i = 0; i < messages.length; i++) {
220            try {
221
222                if (selectorExpression == null) {
223                    c.add(OpenTypeSupport.convert(messages[i]));
224                } else {
225                    ctx.setMessageReference(messages[i]);
226                    if (selectorExpression.matches(ctx)) {
227                        c.add(OpenTypeSupport.convert(messages[i]));
228                    }
229                }
230
231            } catch (Throwable e) {
232                LOG.warn("exception browsing destination", e);
233            }
234        }
235
236        CompositeData rc[] = new CompositeData[c.size()];
237        c.toArray(rc);
238        return rc;
239    }
240
241    /**
242     * Browses the current destination returning a list of messages
243     */
244    @Override
245    public List<Object> browseMessages() throws InvalidSelectorException {
246        return browseMessages(null);
247    }
248
249    /**
250     * Browses the current destination with the given selector returning a list
251     * of messages
252     */
253    @Override
254    public List<Object> browseMessages(String selector) throws InvalidSelectorException {
255        Message[] messages = destination.browse();
256        ArrayList<Object> answer = new ArrayList<Object>();
257
258        MessageEvaluationContext ctx = new MessageEvaluationContext();
259        ctx.setDestination(destination.getActiveMQDestination());
260        BooleanExpression selectorExpression = selector == null ? null : SelectorParser.parse(selector);
261
262        for (int i = 0; i < messages.length; i++) {
263            try {
264                Message message = messages[i];
265                message.setReadOnlyBody(true);
266                if (selectorExpression == null) {
267                    answer.add(message);
268                } else {
269                    ctx.setMessageReference(message);
270                    if (selectorExpression.matches(ctx)) {
271                        answer.add(message);
272                    }
273                }
274
275            } catch (Throwable e) {
276                LOG.warn("exception browsing destination", e);
277            }
278        }
279        return answer;
280    }
281
282    @Override
283    public TabularData browseAsTable() throws OpenDataException {
284        try {
285            return browseAsTable(null);
286        } catch (InvalidSelectorException e) {
287            throw new RuntimeException(e);
288        }
289    }
290
291    @Override
292    public TabularData browseAsTable(String selector) throws OpenDataException, InvalidSelectorException {
293        OpenTypeFactory factory = OpenTypeSupport.getFactory(ActiveMQMessage.class);
294        Message[] messages = destination.browse();
295        CompositeType ct = factory.getCompositeType();
296        TabularType tt = new TabularType("MessageList", "MessageList", ct, new String[] { "JMSMessageID" });
297        TabularDataSupport rc = new TabularDataSupport(tt);
298
299        MessageEvaluationContext ctx = new MessageEvaluationContext();
300        ctx.setDestination(destination.getActiveMQDestination());
301        BooleanExpression selectorExpression = selector == null ? null : SelectorParser.parse(selector);
302
303        for (int i = 0; i < messages.length; i++) {
304            try {
305                if (selectorExpression == null) {
306                    rc.put(new CompositeDataSupport(ct, factory.getFields(messages[i])));
307                } else {
308                    ctx.setMessageReference(messages[i]);
309                    if (selectorExpression.matches(ctx)) {
310                        rc.put(new CompositeDataSupport(ct, factory.getFields(messages[i])));
311                    }
312                }
313            } catch (Throwable e) {
314                LOG.warn("exception browsing destination", e);
315            }
316        }
317
318        return rc;
319    }
320
321    @Override
322    public String sendTextMessageWithProperties(String properties) throws Exception {
323        String[] kvs = properties.split(",");
324        Map<String, String> props = new HashMap<String, String>();
325        for (String kv : kvs) {
326            String[] it = kv.split("=");
327            if (it.length == 2) {
328                props.put(it[0],it[1]);
329            }
330        }
331        return sendTextMessage(props, props.remove("body"), props.remove("username"), props.remove("password"));
332    }
333
334    @Override
335    public String sendTextMessage(String body) throws Exception {
336        return sendTextMessage(Collections.EMPTY_MAP, body);
337    }
338
339    @Override
340    public String sendTextMessage(Map headers, String body) throws Exception {
341        return sendTextMessage(headers, body, null, null);
342    }
343
344    @Override
345    public String sendTextMessage(String body, String user, @Sensitive String password) throws Exception {
346        return sendTextMessage(Collections.EMPTY_MAP, body, user, password);
347    }
348
349    @Override
350    public String sendTextMessage(Map<String, String> headers, String body, String userName, @Sensitive String password) throws Exception {
351
352        String brokerUrl = "vm://" + broker.getBrokerName();
353        ActiveMQDestination dest = destination.getActiveMQDestination();
354
355        ActiveMQConnectionFactory cf = new ActiveMQConnectionFactory(brokerUrl);
356        Connection connection = null;
357        try {
358
359            connection = cf.createConnection(userName, password);
360            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
361            MessageProducer producer = session.createProducer(dest);
362            ActiveMQTextMessage msg = (ActiveMQTextMessage) session.createTextMessage(body);
363
364            for (Iterator iter = headers.entrySet().iterator(); iter.hasNext();) {
365                Map.Entry entry = (Map.Entry) iter.next();
366                msg.setObjectProperty((String) entry.getKey(), entry.getValue());
367            }
368
369            producer.setDeliveryMode(msg.getJMSDeliveryMode());
370            producer.setPriority(msg.getPriority());
371            long ttl = 0;
372            if (msg.getExpiration() != 0) {
373                ttl = msg.getExpiration() - System.currentTimeMillis();
374            } else {
375                String timeToLive = headers.get("timeToLive");
376                if (timeToLive != null) {
377                    ttl = Integer.valueOf(timeToLive);
378                }
379            }
380            producer.setTimeToLive(ttl > 0 ? ttl : 0);
381            producer.send(msg);
382
383            return msg.getJMSMessageID();
384
385        } finally {
386            connection.close();
387        }
388
389    }
390
391    @Override
392    public int getMaxAuditDepth() {
393        return destination.getMaxAuditDepth();
394    }
395
396    @Override
397    public int getMaxProducersToAudit() {
398        return destination.getMaxProducersToAudit();
399    }
400
401    public boolean isEnableAudit() {
402        return destination.isEnableAudit();
403    }
404
405    public void setEnableAudit(boolean enableAudit) {
406        destination.setEnableAudit(enableAudit);
407    }
408
409    @Override
410    public void setMaxAuditDepth(int maxAuditDepth) {
411        destination.setMaxAuditDepth(maxAuditDepth);
412    }
413
414    @Override
415    public void setMaxProducersToAudit(int maxProducersToAudit) {
416        destination.setMaxProducersToAudit(maxProducersToAudit);
417    }
418
419    @Override
420    public float getMemoryUsagePortion() {
421        return destination.getMemoryUsage().getUsagePortion();
422    }
423
424    @Override
425    public long getProducerCount() {
426        return destination.getDestinationStatistics().getProducers().getCount();
427    }
428
429    @Override
430    public boolean isProducerFlowControl() {
431        return destination.isProducerFlowControl();
432    }
433
434    @Override
435    public void setMemoryUsagePortion(float value) {
436        destination.getMemoryUsage().setUsagePortion(value);
437    }
438
439    @Override
440    public void setProducerFlowControl(boolean producerFlowControl) {
441        destination.setProducerFlowControl(producerFlowControl);
442    }
443
444    @Override
445    public boolean isAlwaysRetroactive() {
446        return destination.isAlwaysRetroactive();
447    }
448
449    @Override
450    public void setAlwaysRetroactive(boolean alwaysRetroactive) {
451        destination.setAlwaysRetroactive(alwaysRetroactive);
452    }
453
454    /**
455     * Set's the interval at which warnings about producers being blocked by
456     * resource usage will be triggered. Values of 0 or less will disable
457     * warnings
458     *
459     * @param blockedProducerWarningInterval the interval at which warning about
460     *            blocked producers will be triggered.
461     */
462    @Override
463    public void setBlockedProducerWarningInterval(long blockedProducerWarningInterval) {
464        destination.setBlockedProducerWarningInterval(blockedProducerWarningInterval);
465    }
466
467    /**
468     *
469     * @return the interval at which warning about blocked producers will be
470     *         triggered.
471     */
472    @Override
473    public long getBlockedProducerWarningInterval() {
474        return destination.getBlockedProducerWarningInterval();
475    }
476
477    @Override
478    public int getMaxPageSize() {
479        return destination.getMaxPageSize();
480    }
481
482    @Override
483    public void setMaxPageSize(int pageSize) {
484        destination.setMaxPageSize(pageSize);
485    }
486
487    @Override
488    public boolean isUseCache() {
489        return destination.isUseCache();
490    }
491
492    @Override
493    public void setUseCache(boolean value) {
494        destination.setUseCache(value);
495    }
496
497    @Override
498    public ObjectName[] getSubscriptions() throws IOException, MalformedObjectNameException {
499        List<Subscription> subscriptions = destination.getConsumers();
500        ObjectName[] answer = new ObjectName[subscriptions.size()];
501        ObjectName brokerObjectName = broker.getBrokerService().getBrokerObjectName();
502        int index = 0;
503        for (Subscription subscription : subscriptions) {
504            String connectionClientId = subscription.getContext().getClientId();
505            answer[index++] = BrokerMBeanSupport.createSubscriptionName(brokerObjectName, connectionClientId, subscription.getConsumerInfo());
506        }
507        return answer;
508    }
509
510    @Override
511    public ObjectName getSlowConsumerStrategy() throws IOException, MalformedObjectNameException {
512        ObjectName result = null;
513        SlowConsumerStrategy strategy = destination.getSlowConsumerStrategy();
514        if (strategy != null && strategy instanceof AbortSlowConsumerStrategy) {
515            result = broker.registerSlowConsumerStrategy((AbortSlowConsumerStrategy)strategy);
516        }
517        return result;
518    }
519
520    @Override
521    public String getOptions() {
522        Map<String, String> options = destination.getActiveMQDestination().getOptions();
523        String optionsString = "";
524        try {
525            if (options != null) {
526                optionsString = URISupport.createQueryString(options);
527            }
528        } catch (URISyntaxException ignored) {}
529        return optionsString;
530    }
531
532    @Override
533    public boolean isDLQ() {
534        return destination.getActiveMQDestination().isDLQ();
535    }
536
537    @Override
538    public long getBlockedSends() {
539        return destination.getDestinationStatistics().getBlockedSends().getCount();
540    }
541
542    @Override
543    public double getAverageBlockedTime() {
544        return destination.getDestinationStatistics().getBlockedTime().getAverageTime();
545    }
546
547    @Override
548    public long getTotalBlockedTime() {
549        return destination.getDestinationStatistics().getBlockedTime().getTotalTime();
550    }
551
552}