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}