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.advisory; 018 019import java.util.ArrayList; 020import java.util.Collection; 021import java.util.Collections; 022import java.util.Iterator; 023import java.util.LinkedHashMap; 024import java.util.Map; 025import java.util.Set; 026import java.util.concurrent.ConcurrentHashMap; 027import java.util.concurrent.ConcurrentMap; 028import java.util.concurrent.locks.ReentrantReadWriteLock; 029 030import org.apache.activemq.broker.Broker; 031import org.apache.activemq.broker.BrokerFilter; 032import org.apache.activemq.broker.BrokerService; 033import org.apache.activemq.broker.ConnectionContext; 034import org.apache.activemq.broker.ProducerBrokerExchange; 035import org.apache.activemq.broker.TransportConnector; 036import org.apache.activemq.broker.region.BaseDestination; 037import org.apache.activemq.broker.region.Destination; 038import org.apache.activemq.broker.region.DurableTopicSubscription; 039import org.apache.activemq.broker.region.MessageReference; 040import org.apache.activemq.broker.region.RegionBroker; 041import org.apache.activemq.broker.region.Subscription; 042import org.apache.activemq.broker.region.TopicRegion; 043import org.apache.activemq.broker.region.TopicSubscription; 044import org.apache.activemq.broker.region.virtual.VirtualDestination; 045import org.apache.activemq.command.ActiveMQDestination; 046import org.apache.activemq.command.ActiveMQMessage; 047import org.apache.activemq.command.ActiveMQTopic; 048import org.apache.activemq.command.BrokerInfo; 049import org.apache.activemq.command.Command; 050import org.apache.activemq.command.ConnectionId; 051import org.apache.activemq.command.ConnectionInfo; 052import org.apache.activemq.command.ConsumerId; 053import org.apache.activemq.command.ConsumerInfo; 054import org.apache.activemq.command.DestinationInfo; 055import org.apache.activemq.command.Message; 056import org.apache.activemq.command.MessageId; 057import org.apache.activemq.command.ProducerId; 058import org.apache.activemq.command.ProducerInfo; 059import org.apache.activemq.command.RemoveSubscriptionInfo; 060import org.apache.activemq.command.SessionId; 061import org.apache.activemq.security.SecurityContext; 062import org.apache.activemq.state.ProducerState; 063import org.apache.activemq.usage.Usage; 064import org.apache.activemq.util.IdGenerator; 065import org.apache.activemq.util.LongSequenceGenerator; 066import org.apache.activemq.util.SubscriptionKey; 067import org.slf4j.Logger; 068import org.slf4j.LoggerFactory; 069 070/** 071 * This broker filter handles tracking the state of the broker for purposes of 072 * publishing advisory messages to advisory consumers. 073 */ 074public class AdvisoryBroker extends BrokerFilter { 075 076 private static final Logger LOG = LoggerFactory.getLogger(AdvisoryBroker.class); 077 private static final IdGenerator ID_GENERATOR = new IdGenerator(); 078 079 protected final ConcurrentMap<ConnectionId, ConnectionInfo> connections = new ConcurrentHashMap<ConnectionId, ConnectionInfo>(); 080 081 private final ReentrantReadWriteLock consumersLock = new ReentrantReadWriteLock(); 082 protected final Map<ConsumerId, ConsumerInfo> consumers = new LinkedHashMap<ConsumerId, ConsumerInfo>(); 083 084 /** 085 * This is a set to track all of the virtual destinations that have been added to the broker so 086 * they can be easily referenced later. 087 */ 088 protected final Set<VirtualDestination> virtualDestinations = Collections.newSetFromMap(new ConcurrentHashMap<VirtualDestination, Boolean>()); 089 /** 090 * This is a map to track all consumers that exist on the virtual destination so that we can fire 091 * an advisory later when they go away to remove the demand. 092 */ 093 protected final ConcurrentMap<ConsumerInfo, VirtualDestination> virtualDestinationConsumers = new ConcurrentHashMap<>(); 094 /** 095 * This is a map to track unique demand for the existence of a virtual destination so we make sure 096 * we don't send duplicate advisories. 097 */ 098 protected final ConcurrentMap<VirtualConsumerPair, ConsumerInfo> brokerConsumerDests = new ConcurrentHashMap<>(); 099 100 protected final ConcurrentMap<ProducerId, ProducerInfo> producers = new ConcurrentHashMap<ProducerId, ProducerInfo>(); 101 protected final ConcurrentMap<ActiveMQDestination, DestinationInfo> destinations = new ConcurrentHashMap<ActiveMQDestination, DestinationInfo>(); 102 protected final ConcurrentMap<BrokerInfo, ActiveMQMessage> networkBridges = new ConcurrentHashMap<BrokerInfo, ActiveMQMessage>(); 103 protected final ProducerId advisoryProducerId = new ProducerId(); 104 105 private final LongSequenceGenerator messageIdGenerator = new LongSequenceGenerator(); 106 107 private VirtualDestinationMatcher virtualDestinationMatcher = new DestinationFilterVirtualDestinationMatcher(); 108 109 public AdvisoryBroker(Broker next) { 110 super(next); 111 advisoryProducerId.setConnectionId(ID_GENERATOR.generateId()); 112 } 113 114 @Override 115 public void addConnection(ConnectionContext context, ConnectionInfo info) throws Exception { 116 super.addConnection(context, info); 117 118 ActiveMQTopic topic = AdvisorySupport.getConnectionAdvisoryTopic(); 119 // do not distribute passwords in advisory messages. usernames okay 120 ConnectionInfo copy = info.copy(); 121 copy.setPassword(""); 122 fireAdvisory(context, topic, copy); 123 connections.put(copy.getConnectionId(), copy); 124 } 125 126 @Override 127 public Subscription addConsumer(ConnectionContext context, ConsumerInfo info) throws Exception { 128 Subscription answer = super.addConsumer(context, info); 129 130 // Don't advise advisory topics. 131 if (!AdvisorySupport.isAdvisoryTopic(info.getDestination())) { 132 ActiveMQTopic topic = AdvisorySupport.getConsumerAdvisoryTopic(info.getDestination()); 133 consumersLock.writeLock().lock(); 134 try { 135 consumers.put(info.getConsumerId(), info); 136 137 //check if this is a consumer on a destination that matches a virtual destination 138 if (getBrokerService().isUseVirtualDestSubs()) { 139 for (VirtualDestination virtualDestination : virtualDestinations) { 140 if (virtualDestinationMatcher.matches(virtualDestination, info.getDestination())) { 141 fireVirtualDestinationAddAdvisory(context, info, info.getDestination(), virtualDestination); 142 } 143 } 144 } 145 } finally { 146 consumersLock.writeLock().unlock(); 147 } 148 fireConsumerAdvisory(context, info.getDestination(), topic, info); 149 } else { 150 // We need to replay all the previously collected state objects 151 // for this newly added consumer. 152 if (AdvisorySupport.isConnectionAdvisoryTopic(info.getDestination())) { 153 // Replay the connections. 154 for (Iterator<ConnectionInfo> iter = connections.values().iterator(); iter.hasNext(); ) { 155 ConnectionInfo value = iter.next(); 156 ActiveMQTopic topic = AdvisorySupport.getConnectionAdvisoryTopic(); 157 fireAdvisory(context, topic, value, info.getConsumerId()); 158 } 159 } 160 161 // We check here whether the Destination is Temporary Destination specific or not since we 162 // can avoid sending advisory messages to the consumer if it only wants Temporary Destination 163 // notifications. If its not just temporary destination related destinations then we have 164 // to send them all, a composite destination could want both. 165 if (AdvisorySupport.isTempDestinationAdvisoryTopic(info.getDestination())) { 166 // Replay the temporary destinations. 167 for (DestinationInfo destination : destinations.values()) { 168 if (destination.getDestination().isTemporary()) { 169 ActiveMQTopic topic = AdvisorySupport.getDestinationAdvisoryTopic(destination.getDestination()); 170 fireAdvisory(context, topic, destination, info.getConsumerId()); 171 } 172 } 173 } else if (AdvisorySupport.isDestinationAdvisoryTopic(info.getDestination())) { 174 // Replay all the destinations. 175 for (DestinationInfo destination : destinations.values()) { 176 ActiveMQTopic topic = AdvisorySupport.getDestinationAdvisoryTopic(destination.getDestination()); 177 fireAdvisory(context, topic, destination, info.getConsumerId()); 178 } 179 } 180 181 // Replay the producers. 182 if (AdvisorySupport.isProducerAdvisoryTopic(info.getDestination())) { 183 for (Iterator<ProducerInfo> iter = producers.values().iterator(); iter.hasNext(); ) { 184 ProducerInfo value = iter.next(); 185 ActiveMQTopic topic = AdvisorySupport.getProducerAdvisoryTopic(value.getDestination()); 186 fireProducerAdvisory(context, value.getDestination(), topic, value, info.getConsumerId()); 187 } 188 } 189 190 // Replay the consumers. 191 if (AdvisorySupport.isConsumerAdvisoryTopic(info.getDestination())) { 192 consumersLock.readLock().lock(); 193 try { 194 for (Iterator<ConsumerInfo> iter = consumers.values().iterator(); iter.hasNext(); ) { 195 ConsumerInfo value = iter.next(); 196 ActiveMQTopic topic = AdvisorySupport.getConsumerAdvisoryTopic(value.getDestination()); 197 fireConsumerAdvisory(context, value.getDestination(), topic, value, info.getConsumerId()); 198 } 199 } finally { 200 consumersLock.readLock().unlock(); 201 } 202 } 203 204 // Replay the virtual destination consumers. 205 if (AdvisorySupport.isVirtualDestinationConsumerAdvisoryTopic(info.getDestination())) { 206 for (Iterator<ConsumerInfo> iter = virtualDestinationConsumers.keySet().iterator(); iter.hasNext(); ) { 207 ConsumerInfo key = iter.next(); 208 ActiveMQTopic topic = AdvisorySupport.getVirtualDestinationConsumerAdvisoryTopic(key.getDestination()); 209 fireConsumerAdvisory(context, key.getDestination(), topic, key); 210 } 211 } 212 213 // Replay network bridges 214 if (AdvisorySupport.isNetworkBridgeAdvisoryTopic(info.getDestination())) { 215 for (Iterator<BrokerInfo> iter = networkBridges.keySet().iterator(); iter.hasNext(); ) { 216 BrokerInfo key = iter.next(); 217 ActiveMQTopic topic = AdvisorySupport.getNetworkBridgeAdvisoryTopic(); 218 fireAdvisory(context, topic, key, null, networkBridges.get(key)); 219 } 220 } 221 } 222 return answer; 223 } 224 225 @Override 226 public void addProducer(ConnectionContext context, ProducerInfo info) throws Exception { 227 super.addProducer(context, info); 228 229 // Don't advise advisory topics. 230 if (info.getDestination() != null && !AdvisorySupport.isAdvisoryTopic(info.getDestination())) { 231 ActiveMQTopic topic = AdvisorySupport.getProducerAdvisoryTopic(info.getDestination()); 232 fireProducerAdvisory(context, info.getDestination(), topic, info); 233 producers.put(info.getProducerId(), info); 234 } 235 } 236 237 @Override 238 public Destination addDestination(ConnectionContext context, ActiveMQDestination destination, boolean create) throws Exception { 239 Destination answer = super.addDestination(context, destination, create); 240 if (!AdvisorySupport.isAdvisoryTopic(destination)) { 241 //for queues, create demand if isUseVirtualDestSubsOnCreation is true 242 if (getBrokerService().isUseVirtualDestSubsOnCreation() && destination.isQueue()) { 243 //check if this new destination matches a virtual destination that exists 244 for (VirtualDestination virtualDestination : virtualDestinations) { 245 if (virtualDestinationMatcher.matches(virtualDestination, destination)) { 246 fireVirtualDestinationAddAdvisory(context, null, destination, virtualDestination); 247 } 248 } 249 } 250 251 DestinationInfo info = new DestinationInfo(context.getConnectionId(), DestinationInfo.ADD_OPERATION_TYPE, destination); 252 DestinationInfo previous = destinations.putIfAbsent(destination, info); 253 if (previous == null) { 254 ActiveMQTopic topic = AdvisorySupport.getDestinationAdvisoryTopic(destination); 255 fireAdvisory(context, topic, info); 256 } 257 } 258 return answer; 259 } 260 261 @Override 262 public void addDestinationInfo(ConnectionContext context, DestinationInfo info) throws Exception { 263 ActiveMQDestination destination = info.getDestination(); 264 next.addDestinationInfo(context, info); 265 266 if (!AdvisorySupport.isAdvisoryTopic(destination)) { 267 DestinationInfo previous = destinations.putIfAbsent(destination, info); 268 if (previous == null) { 269 ActiveMQTopic topic = AdvisorySupport.getDestinationAdvisoryTopic(destination); 270 fireAdvisory(context, topic, info); 271 } 272 } 273 } 274 275 @Override 276 public void removeDestination(ConnectionContext context, ActiveMQDestination destination, long timeout) throws Exception { 277 super.removeDestination(context, destination, timeout); 278 DestinationInfo info = destinations.remove(destination); 279 if (info != null) { 280 281 //on destination removal, remove all demand if using virtual dest subs 282 if (getBrokerService().isUseVirtualDestSubs()) { 283 for (ConsumerInfo consumerInfo : virtualDestinationConsumers.keySet()) { 284 //find all consumers for this virtual destination 285 VirtualDestination virtualDestination = virtualDestinationConsumers.get(consumerInfo); 286 287 //find a consumer that matches this virtualDest and destination 288 if (virtualDestinationMatcher.matches(virtualDestination, destination)) { 289 //in case of multiple matches 290 VirtualConsumerPair key = new VirtualConsumerPair(virtualDestination, destination); 291 ConsumerInfo i = brokerConsumerDests.get(key); 292 if (consumerInfo.equals(i)) { 293 if (brokerConsumerDests.remove(key) != null) { 294 fireVirtualDestinationRemoveAdvisory(context, consumerInfo); 295 break; 296 } 297 } 298 } 299 } 300 } 301 302 // ensure we don't modify (and loose/overwrite) an in-flight add advisory, so duplicate 303 info = info.copy(); 304 info.setDestination(destination); 305 info.setOperationType(DestinationInfo.REMOVE_OPERATION_TYPE); 306 ActiveMQTopic topic = AdvisorySupport.getDestinationAdvisoryTopic(destination); 307 fireAdvisory(context, topic, info); 308 ActiveMQTopic[] advisoryDestinations = AdvisorySupport.getAllDestinationAdvisoryTopics(destination); 309 for (ActiveMQTopic advisoryDestination : advisoryDestinations) { 310 try { 311 next.removeDestination(context, advisoryDestination, -1); 312 } catch (Exception expectedIfDestinationDidNotExistYet) { 313 } 314 } 315 } 316 } 317 318 @Override 319 public void removeDestinationInfo(ConnectionContext context, DestinationInfo destInfo) throws Exception { 320 super.removeDestinationInfo(context, destInfo); 321 DestinationInfo info = destinations.remove(destInfo.getDestination()); 322 if (info != null) { 323 // ensure we don't modify (and loose/overwrite) an in-flight add advisory, so duplicate 324 info = info.copy(); 325 info.setDestination(destInfo.getDestination()); 326 info.setOperationType(DestinationInfo.REMOVE_OPERATION_TYPE); 327 ActiveMQTopic topic = AdvisorySupport.getDestinationAdvisoryTopic(destInfo.getDestination()); 328 fireAdvisory(context, topic, info); 329 ActiveMQTopic[] advisoryDestinations = AdvisorySupport.getAllDestinationAdvisoryTopics(destInfo.getDestination()); 330 for (ActiveMQTopic advisoryDestination : advisoryDestinations) { 331 try { 332 next.removeDestination(context, advisoryDestination, -1); 333 } catch (Exception expectedIfDestinationDidNotExistYet) { 334 } 335 } 336 } 337 } 338 339 @Override 340 public void removeConnection(ConnectionContext context, ConnectionInfo info, Throwable error) throws Exception { 341 super.removeConnection(context, info, error); 342 343 ActiveMQTopic topic = AdvisorySupport.getConnectionAdvisoryTopic(); 344 fireAdvisory(context, topic, info.createRemoveCommand()); 345 connections.remove(info.getConnectionId()); 346 } 347 348 @Override 349 public void removeConsumer(ConnectionContext context, ConsumerInfo info) throws Exception { 350 super.removeConsumer(context, info); 351 352 // Don't advise advisory topics. 353 ActiveMQDestination dest = info.getDestination(); 354 if (!AdvisorySupport.isAdvisoryTopic(dest)) { 355 ActiveMQTopic topic = AdvisorySupport.getConsumerAdvisoryTopic(dest); 356 consumersLock.writeLock().lock(); 357 try { 358 consumers.remove(info.getConsumerId()); 359 360 //remove the demand for this consumer if it matches a virtual destination 361 if(getBrokerService().isUseVirtualDestSubs()) { 362 fireVirtualDestinationRemoveAdvisory(context, info); 363 } 364 } finally { 365 consumersLock.writeLock().unlock(); 366 } 367 if (!dest.isTemporary() || destinations.containsKey(dest)) { 368 fireConsumerAdvisory(context, dest, topic, info.createRemoveCommand()); 369 } 370 } 371 } 372 373 @Override 374 public void removeSubscription(ConnectionContext context, RemoveSubscriptionInfo info) throws Exception { 375 SubscriptionKey key = new SubscriptionKey(context.getClientId(), info.getSubscriptionName()); 376 377 RegionBroker regionBroker = null; 378 if (next instanceof RegionBroker) { 379 regionBroker = (RegionBroker) next; 380 } else { 381 BrokerService service = next.getBrokerService(); 382 regionBroker = (RegionBroker) service.getRegionBroker(); 383 } 384 385 if (regionBroker == null) { 386 LOG.warn("Cannot locate a RegionBroker instance to pass along the removeSubscription call"); 387 throw new IllegalStateException("No RegionBroker found."); 388 } 389 390 DurableTopicSubscription sub = ((TopicRegion) regionBroker.getTopicRegion()).getDurableSubscription(key); 391 392 super.removeSubscription(context, info); 393 394 if (sub == null) { 395 LOG.warn("We cannot send an advisory message for a durable sub removal when we don't know about the durable sub"); 396 return; 397 } 398 399 ActiveMQDestination dest = sub.getConsumerInfo().getDestination(); 400 401 // Don't advise advisory topics. 402 if (!AdvisorySupport.isAdvisoryTopic(dest)) { 403 ActiveMQTopic topic = AdvisorySupport.getConsumerAdvisoryTopic(dest); 404 fireConsumerAdvisory(context, dest, topic, info); 405 } 406 407 } 408 409 @Override 410 public void removeProducer(ConnectionContext context, ProducerInfo info) throws Exception { 411 super.removeProducer(context, info); 412 413 // Don't advise advisory topics. 414 ActiveMQDestination dest = info.getDestination(); 415 if (info.getDestination() != null && !AdvisorySupport.isAdvisoryTopic(dest)) { 416 ActiveMQTopic topic = AdvisorySupport.getProducerAdvisoryTopic(dest); 417 producers.remove(info.getProducerId()); 418 if (!dest.isTemporary() || destinations.containsKey(dest)) { 419 fireProducerAdvisory(context, dest, topic, info.createRemoveCommand()); 420 } 421 } 422 } 423 424 @Override 425 public void messageExpired(ConnectionContext context, MessageReference messageReference, Subscription subscription) { 426 super.messageExpired(context, messageReference, subscription); 427 try { 428 if (!messageReference.isAdvisory()) { 429 BaseDestination baseDestination = (BaseDestination) messageReference.getMessage().getRegionDestination(); 430 ActiveMQTopic topic = AdvisorySupport.getExpiredMessageTopic(baseDestination.getActiveMQDestination()); 431 Message payload = messageReference.getMessage().copy(); 432 if (!baseDestination.isIncludeBodyForAdvisory()) { 433 payload.clearBody(); 434 } 435 ActiveMQMessage advisoryMessage = new ActiveMQMessage(); 436 advisoryMessage.setStringProperty(AdvisorySupport.MSG_PROPERTY_MESSAGE_ID, payload.getMessageId().toString()); 437 fireAdvisory(context, topic, payload, null, advisoryMessage); 438 } 439 } catch (Exception e) { 440 handleFireFailure("expired", e); 441 } 442 } 443 444 @Override 445 public void messageConsumed(ConnectionContext context, MessageReference messageReference) { 446 super.messageConsumed(context, messageReference); 447 try { 448 if (!messageReference.isAdvisory()) { 449 BaseDestination baseDestination = (BaseDestination) messageReference.getMessage().getRegionDestination(); 450 ActiveMQTopic topic = AdvisorySupport.getMessageConsumedAdvisoryTopic(baseDestination.getActiveMQDestination()); 451 Message payload = messageReference.getMessage().copy(); 452 if (!baseDestination.isIncludeBodyForAdvisory()) { 453 payload.clearBody(); 454 } 455 ActiveMQMessage advisoryMessage = new ActiveMQMessage(); 456 advisoryMessage.setStringProperty(AdvisorySupport.MSG_PROPERTY_MESSAGE_ID, payload.getMessageId().toString()); 457 advisoryMessage.setStringProperty(AdvisorySupport.MSG_PROPERTY_DESTINATION, baseDestination.getActiveMQDestination().getQualifiedName()); 458 fireAdvisory(context, topic, payload, null, advisoryMessage); 459 } 460 } catch (Exception e) { 461 handleFireFailure("consumed", e); 462 } 463 } 464 465 @Override 466 public void messageDelivered(ConnectionContext context, MessageReference messageReference) { 467 super.messageDelivered(context, messageReference); 468 try { 469 if (!messageReference.isAdvisory()) { 470 BaseDestination baseDestination = (BaseDestination) messageReference.getMessage().getRegionDestination(); 471 ActiveMQTopic topic = AdvisorySupport.getMessageDeliveredAdvisoryTopic(baseDestination.getActiveMQDestination()); 472 Message payload = messageReference.getMessage().copy(); 473 if (!baseDestination.isIncludeBodyForAdvisory()) { 474 payload.clearBody(); 475 } 476 ActiveMQMessage advisoryMessage = new ActiveMQMessage(); 477 advisoryMessage.setStringProperty(AdvisorySupport.MSG_PROPERTY_MESSAGE_ID, payload.getMessageId().toString()); 478 advisoryMessage.setStringProperty(AdvisorySupport.MSG_PROPERTY_DESTINATION, baseDestination.getActiveMQDestination().getQualifiedName()); 479 fireAdvisory(context, topic, payload, null, advisoryMessage); 480 } 481 } catch (Exception e) { 482 handleFireFailure("delivered", e); 483 } 484 } 485 486 @Override 487 public void messageDiscarded(ConnectionContext context, Subscription sub, MessageReference messageReference) { 488 super.messageDiscarded(context, sub, messageReference); 489 try { 490 if (!messageReference.isAdvisory()) { 491 BaseDestination baseDestination = (BaseDestination) messageReference.getMessage().getRegionDestination(); 492 ActiveMQTopic topic = AdvisorySupport.getMessageDiscardedAdvisoryTopic(baseDestination.getActiveMQDestination()); 493 Message payload = messageReference.getMessage().copy(); 494 if (!baseDestination.isIncludeBodyForAdvisory()) { 495 payload.clearBody(); 496 } 497 ActiveMQMessage advisoryMessage = new ActiveMQMessage(); 498 if (sub instanceof TopicSubscription) { 499 advisoryMessage.setIntProperty(AdvisorySupport.MSG_PROPERTY_DISCARDED_COUNT, ((TopicSubscription) sub).discarded()); 500 } 501 advisoryMessage.setStringProperty(AdvisorySupport.MSG_PROPERTY_MESSAGE_ID, payload.getMessageId().toString()); 502 advisoryMessage.setStringProperty(AdvisorySupport.MSG_PROPERTY_CONSUMER_ID, sub.getConsumerInfo().getConsumerId().toString()); 503 advisoryMessage.setStringProperty(AdvisorySupport.MSG_PROPERTY_DESTINATION, baseDestination.getActiveMQDestination().getQualifiedName()); 504 505 fireAdvisory(context, topic, payload, null, advisoryMessage); 506 } 507 } catch (Exception e) { 508 handleFireFailure("discarded", e); 509 } 510 } 511 512 @Override 513 public void slowConsumer(ConnectionContext context, Destination destination, Subscription subs) { 514 super.slowConsumer(context, destination, subs); 515 try { 516 if (!AdvisorySupport.isAdvisoryTopic(destination.getActiveMQDestination())) { 517 ActiveMQTopic topic = AdvisorySupport.getSlowConsumerAdvisoryTopic(destination.getActiveMQDestination()); 518 ActiveMQMessage advisoryMessage = new ActiveMQMessage(); 519 advisoryMessage.setStringProperty(AdvisorySupport.MSG_PROPERTY_CONSUMER_ID, subs.getConsumerInfo().getConsumerId().toString()); 520 fireAdvisory(context, topic, subs.getConsumerInfo(), null, advisoryMessage); 521 } 522 } catch (Exception e) { 523 handleFireFailure("slow consumer", e); 524 } 525 } 526 527 @Override 528 public void fastProducer(ConnectionContext context, ProducerInfo producerInfo, ActiveMQDestination destination) { 529 super.fastProducer(context, producerInfo, destination); 530 try { 531 if (!AdvisorySupport.isAdvisoryTopic(destination)) { 532 ActiveMQTopic topic = AdvisorySupport.getFastProducerAdvisoryTopic(destination); 533 ActiveMQMessage advisoryMessage = new ActiveMQMessage(); 534 advisoryMessage.setStringProperty(AdvisorySupport.MSG_PROPERTY_PRODUCER_ID, producerInfo.getProducerId().toString()); 535 fireAdvisory(context, topic, producerInfo, null, advisoryMessage); 536 } 537 } catch (Exception e) { 538 handleFireFailure("fast producer", e); 539 } 540 } 541 542 private final IdGenerator connectionIdGenerator = new IdGenerator("advisory"); 543 private final LongSequenceGenerator sessionIdGenerator = new LongSequenceGenerator(); 544 private final LongSequenceGenerator consumerIdGenerator = new LongSequenceGenerator(); 545 546 @Override 547 public void virtualDestinationAdded(ConnectionContext context, 548 VirtualDestination virtualDestination) { 549 super.virtualDestinationAdded(context, virtualDestination); 550 551 if (virtualDestinations.add(virtualDestination)) { 552 try { 553 // Don't advise advisory topics. 554 if (!AdvisorySupport.isAdvisoryTopic(virtualDestination.getVirtualDestination())) { 555 556 //create demand for consumers on virtual destinations 557 consumersLock.readLock().lock(); 558 try { 559 //loop through existing destinations to see if any match this newly 560 //created virtual destination 561 if (getBrokerService().isUseVirtualDestSubsOnCreation()) { 562 //for matches that are a queue, fire an advisory for demand 563 for (ActiveMQDestination destination : destinations.keySet()) { 564 if(destination.isQueue()) { 565 if (virtualDestinationMatcher.matches(virtualDestination, destination)) { 566 fireVirtualDestinationAddAdvisory(context, null, destination, virtualDestination); 567 } 568 } 569 } 570 } 571 572 //loop through existing consumers to see if any of them are consuming on a destination 573 //that matches the new virtual destination 574 for (Iterator<ConsumerInfo> iter = consumers.values().iterator(); iter.hasNext(); ) { 575 ConsumerInfo info = iter.next(); 576 if (virtualDestinationMatcher.matches(virtualDestination, info.getDestination())) { 577 fireVirtualDestinationAddAdvisory(context, info, info.getDestination(), virtualDestination); 578 } 579 } 580 } finally { 581 consumersLock.readLock().unlock(); 582 } 583 } 584 } catch (Exception e) { 585 handleFireFailure("virtualDestinationAdded", e); 586 } 587 } 588 } 589 590 private void fireVirtualDestinationAddAdvisory(ConnectionContext context, ConsumerInfo info, ActiveMQDestination activeMQDest, 591 VirtualDestination virtualDestination) throws Exception { 592 //if no consumer info, we need to create one - this is the case when an advisory is fired 593 //because of the existence of a destination matching a virtual destination 594 if (info == null) { 595 ConnectionId connectionId = new ConnectionId(connectionIdGenerator.generateId()); 596 SessionId sessionId = new SessionId(connectionId, sessionIdGenerator.getNextSequenceId()); 597 ConsumerId consumerId = new ConsumerId(sessionId, consumerIdGenerator.getNextSequenceId()); 598 599 info = new ConsumerInfo(consumerId); 600 601 //store the virtual destination and the activeMQDestination as a pair so that we can keep track 602 //of all matching forwarded destinations that caused demand 603 if(brokerConsumerDests.putIfAbsent(new VirtualConsumerPair(virtualDestination, activeMQDest), info) == null) { 604 info.setDestination(virtualDestination.getVirtualDestination()); 605 ActiveMQTopic topic = AdvisorySupport.getVirtualDestinationConsumerAdvisoryTopic(info.getDestination()); 606 607 if (virtualDestinationConsumers.putIfAbsent(info, virtualDestination) == null) { 608 fireConsumerAdvisory(context, info.getDestination(), topic, info); 609 } 610 } 611 //this is the case of a real consumer coming online 612 } else { 613 info = info.copy(); 614 info.setDestination(virtualDestination.getVirtualDestination()); 615 ActiveMQTopic topic = AdvisorySupport.getVirtualDestinationConsumerAdvisoryTopic(info.getDestination()); 616 617 if (virtualDestinationConsumers.putIfAbsent(info, virtualDestination) == null) { 618 fireConsumerAdvisory(context, info.getDestination(), topic, info); 619 } 620 } 621 } 622 623 @Override 624 public void virtualDestinationRemoved(ConnectionContext context, 625 VirtualDestination virtualDestination) { 626 super.virtualDestinationRemoved(context, virtualDestination); 627 628 if (virtualDestinations.remove(virtualDestination)) { 629 try { 630 consumersLock.readLock().lock(); 631 try { 632 // remove the demand created by the addition of the virtual destination 633 if (getBrokerService().isUseVirtualDestSubsOnCreation()) { 634 if (!AdvisorySupport.isAdvisoryTopic(virtualDestination.getVirtualDestination())) { 635 for (ConsumerInfo info : virtualDestinationConsumers.keySet()) { 636 //find all consumers for this virtual destination 637 if (virtualDestinationConsumers.get(info).equals(virtualDestination)) { 638 fireVirtualDestinationRemoveAdvisory(context, info); 639 } 640 641 //check consumers created for the existence of a destination to see if they 642 //match the consumerinfo and clean up 643 for (VirtualConsumerPair activeMQDest : brokerConsumerDests.keySet()) { 644 ConsumerInfo i = brokerConsumerDests.get(activeMQDest); 645 if (info.equals(i)) { 646 brokerConsumerDests.remove(activeMQDest); 647 } 648 } 649 } 650 } 651 } 652 } finally { 653 consumersLock.readLock().unlock(); 654 } 655 } catch (Exception e) { 656 handleFireFailure("virtualDestinationAdded", e); 657 } 658 } 659 } 660 661 private void fireVirtualDestinationRemoveAdvisory(ConnectionContext context, 662 ConsumerInfo info) throws Exception { 663 664 VirtualDestination virtualDestination = virtualDestinationConsumers.remove(info); 665 if (virtualDestination != null) { 666 ActiveMQTopic topic = AdvisorySupport.getVirtualDestinationConsumerAdvisoryTopic(virtualDestination.getVirtualDestination()); 667 668 ActiveMQDestination dest = info.getDestination(); 669 670 if (!dest.isTemporary() || destinations.containsKey(dest)) { 671 fireConsumerAdvisory(context, dest, topic, info.createRemoveCommand()); 672 } 673 } 674 } 675 676 @Override 677 public void isFull(ConnectionContext context, Destination destination, Usage usage) { 678 super.isFull(context, destination, usage); 679 if (AdvisorySupport.isAdvisoryTopic(destination.getActiveMQDestination()) == false) { 680 try { 681 682 ActiveMQTopic topic = AdvisorySupport.getFullAdvisoryTopic(destination.getActiveMQDestination()); 683 ActiveMQMessage advisoryMessage = new ActiveMQMessage(); 684 advisoryMessage.setStringProperty(AdvisorySupport.MSG_PROPERTY_USAGE_NAME, usage.getName()); 685 fireAdvisory(context, topic, null, null, advisoryMessage); 686 687 } catch (Exception e) { 688 handleFireFailure("is full", e); 689 } 690 } 691 } 692 693 @Override 694 public void nowMasterBroker() { 695 super.nowMasterBroker(); 696 try { 697 ActiveMQTopic topic = AdvisorySupport.getMasterBrokerAdvisoryTopic(); 698 ActiveMQMessage advisoryMessage = new ActiveMQMessage(); 699 ConnectionContext context = new ConnectionContext(); 700 context.setSecurityContext(SecurityContext.BROKER_SECURITY_CONTEXT); 701 context.setBroker(getBrokerService().getBroker()); 702 fireAdvisory(context, topic, null, null, advisoryMessage); 703 } catch (Exception e) { 704 handleFireFailure("now master broker", e); 705 } 706 } 707 708 @Override 709 public boolean sendToDeadLetterQueue(ConnectionContext context, MessageReference messageReference, 710 Subscription subscription, Throwable poisonCause) { 711 boolean wasDLQd = super.sendToDeadLetterQueue(context, messageReference, subscription, poisonCause); 712 if (wasDLQd) { 713 try { 714 if (!messageReference.isAdvisory()) { 715 BaseDestination baseDestination = (BaseDestination) messageReference.getMessage().getRegionDestination(); 716 ActiveMQTopic topic = AdvisorySupport.getMessageDLQdAdvisoryTopic(baseDestination.getActiveMQDestination()); 717 Message payload = messageReference.getMessage().copy(); 718 if (!baseDestination.isIncludeBodyForAdvisory()) { 719 payload.clearBody(); 720 } 721 fireAdvisory(context, topic, payload); 722 } 723 } catch (Exception e) { 724 handleFireFailure("add to DLQ", e); 725 } 726 } 727 728 return wasDLQd; 729 } 730 731 @Override 732 public void networkBridgeStarted(BrokerInfo brokerInfo, boolean createdByDuplex, String remoteIp) { 733 try { 734 if (brokerInfo != null) { 735 ActiveMQMessage advisoryMessage = new ActiveMQMessage(); 736 advisoryMessage.setBooleanProperty("started", true); 737 advisoryMessage.setBooleanProperty("createdByDuplex", createdByDuplex); 738 advisoryMessage.setStringProperty("remoteIp", remoteIp); 739 networkBridges.putIfAbsent(brokerInfo, advisoryMessage); 740 741 ActiveMQTopic topic = AdvisorySupport.getNetworkBridgeAdvisoryTopic(); 742 743 ConnectionContext context = new ConnectionContext(); 744 context.setSecurityContext(SecurityContext.BROKER_SECURITY_CONTEXT); 745 context.setBroker(getBrokerService().getBroker()); 746 fireAdvisory(context, topic, brokerInfo, null, advisoryMessage); 747 } 748 } catch (Exception e) { 749 handleFireFailure("network bridge started", e); 750 } 751 } 752 753 @Override 754 public void networkBridgeStopped(BrokerInfo brokerInfo) { 755 try { 756 if (brokerInfo != null) { 757 ActiveMQMessage advisoryMessage = new ActiveMQMessage(); 758 advisoryMessage.setBooleanProperty("started", false); 759 networkBridges.remove(brokerInfo); 760 761 ActiveMQTopic topic = AdvisorySupport.getNetworkBridgeAdvisoryTopic(); 762 763 ConnectionContext context = new ConnectionContext(); 764 context.setSecurityContext(SecurityContext.BROKER_SECURITY_CONTEXT); 765 context.setBroker(getBrokerService().getBroker()); 766 fireAdvisory(context, topic, brokerInfo, null, advisoryMessage); 767 } 768 } catch (Exception e) { 769 handleFireFailure("network bridge stopped", e); 770 } 771 } 772 773 private void handleFireFailure(String message, Throwable cause) { 774 LOG.warn("Failed to fire {} advisory, reason: {}", message, cause); 775 LOG.debug("{} detail: {}", message, cause, cause); 776 } 777 778 protected void fireAdvisory(ConnectionContext context, ActiveMQTopic topic, Command command) throws Exception { 779 fireAdvisory(context, topic, command, null); 780 } 781 782 protected void fireAdvisory(ConnectionContext context, ActiveMQTopic topic, Command command, ConsumerId targetConsumerId) throws Exception { 783 ActiveMQMessage advisoryMessage = new ActiveMQMessage(); 784 fireAdvisory(context, topic, command, targetConsumerId, advisoryMessage); 785 } 786 787 protected void fireConsumerAdvisory(ConnectionContext context, ActiveMQDestination consumerDestination, ActiveMQTopic topic, Command command) throws Exception { 788 fireConsumerAdvisory(context, consumerDestination, topic, command, null); 789 } 790 791 protected void fireConsumerAdvisory(ConnectionContext context, ActiveMQDestination consumerDestination, ActiveMQTopic topic, Command command, ConsumerId targetConsumerId) throws Exception { 792 ActiveMQMessage advisoryMessage = new ActiveMQMessage(); 793 int count = 0; 794 Set<Destination> set = getDestinations(consumerDestination); 795 if (set != null) { 796 for (Destination dest : set) { 797 count += dest.getDestinationStatistics().getConsumers().getCount(); 798 } 799 } 800 advisoryMessage.setIntProperty(AdvisorySupport.MSG_PROPERTY_CONSUMER_COUNT, count); 801 802 fireAdvisory(context, topic, command, targetConsumerId, advisoryMessage); 803 } 804 805 protected void fireProducerAdvisory(ConnectionContext context, ActiveMQDestination producerDestination, ActiveMQTopic topic, Command command) throws Exception { 806 fireProducerAdvisory(context, producerDestination, topic, command, null); 807 } 808 809 protected void fireProducerAdvisory(ConnectionContext context, ActiveMQDestination producerDestination, ActiveMQTopic topic, Command command, ConsumerId targetConsumerId) throws Exception { 810 ActiveMQMessage advisoryMessage = new ActiveMQMessage(); 811 int count = 0; 812 if (producerDestination != null) { 813 Set<Destination> set = getDestinations(producerDestination); 814 if (set != null) { 815 for (Destination dest : set) { 816 count += dest.getDestinationStatistics().getProducers().getCount(); 817 } 818 } 819 } 820 advisoryMessage.setIntProperty("producerCount", count); 821 fireAdvisory(context, topic, command, targetConsumerId, advisoryMessage); 822 } 823 824 public void fireAdvisory(ConnectionContext context, ActiveMQTopic topic, Command command, ConsumerId targetConsumerId, ActiveMQMessage advisoryMessage) throws Exception { 825 if (getBrokerService().isStarted()) { 826 //set properties 827 advisoryMessage.setStringProperty(AdvisorySupport.MSG_PROPERTY_ORIGIN_BROKER_NAME, getBrokerName()); 828 String id = getBrokerId() != null ? getBrokerId().getValue() : "NOT_SET"; 829 advisoryMessage.setStringProperty(AdvisorySupport.MSG_PROPERTY_ORIGIN_BROKER_ID, id); 830 831 String url = getBrokerService().getVmConnectorURI().toString(); 832 //try and find the URL on the transport connector and use if it exists else 833 //try and find a default URL 834 if (context.getConnector() instanceof TransportConnector 835 && ((TransportConnector) context.getConnector()).getPublishableConnectString() != null) { 836 url = ((TransportConnector) context.getConnector()).getPublishableConnectString(); 837 } else if (getBrokerService().getDefaultSocketURIString() != null) { 838 url = getBrokerService().getDefaultSocketURIString(); 839 } 840 advisoryMessage.setStringProperty(AdvisorySupport.MSG_PROPERTY_ORIGIN_BROKER_URL, url); 841 842 //set the data structure 843 advisoryMessage.setDataStructure(command); 844 advisoryMessage.setPersistent(false); 845 advisoryMessage.setType(AdvisorySupport.ADIVSORY_MESSAGE_TYPE); 846 advisoryMessage.setMessageId(new MessageId(advisoryProducerId, messageIdGenerator.getNextSequenceId())); 847 advisoryMessage.setTargetConsumerId(targetConsumerId); 848 advisoryMessage.setDestination(topic); 849 advisoryMessage.setResponseRequired(false); 850 advisoryMessage.setProducerId(advisoryProducerId); 851 boolean originalFlowControl = context.isProducerFlowControl(); 852 final ProducerBrokerExchange producerExchange = new ProducerBrokerExchange(); 853 producerExchange.setConnectionContext(context); 854 producerExchange.setMutable(true); 855 producerExchange.setProducerState(new ProducerState(new ProducerInfo())); 856 try { 857 context.setProducerFlowControl(false); 858 next.send(producerExchange, advisoryMessage); 859 } finally { 860 context.setProducerFlowControl(originalFlowControl); 861 } 862 } 863 } 864 865 public Map<ConnectionId, ConnectionInfo> getAdvisoryConnections() { 866 return connections; 867 } 868 869 public Collection<ConsumerInfo> getAdvisoryConsumers() { 870 consumersLock.readLock().lock(); 871 try { 872 return new ArrayList<ConsumerInfo>(consumers.values()); 873 } finally { 874 consumersLock.readLock().unlock(); 875 } 876 } 877 878 public Map<ProducerId, ProducerInfo> getAdvisoryProducers() { 879 return producers; 880 } 881 882 public Map<ActiveMQDestination, DestinationInfo> getAdvisoryDestinations() { 883 return destinations; 884 } 885 886 private class VirtualConsumerPair { 887 private final VirtualDestination virtualDestination; 888 889 //destination that matches this virtualDestination as part target 890 //this is so we can keep track of more than one destination that might 891 //match the virtualDestination and cause demand 892 private final ActiveMQDestination activeMQDestination; 893 894 public VirtualConsumerPair(VirtualDestination virtualDestination, 895 ActiveMQDestination activeMQDestination) { 896 super(); 897 this.virtualDestination = virtualDestination; 898 this.activeMQDestination = activeMQDestination; 899 } 900 @Override 901 public int hashCode() { 902 final int prime = 31; 903 int result = 1; 904 result = prime * result + getOuterType().hashCode(); 905 result = prime 906 * result 907 + ((activeMQDestination == null) ? 0 : activeMQDestination 908 .hashCode()); 909 result = prime 910 * result 911 + ((virtualDestination == null) ? 0 : virtualDestination 912 .hashCode()); 913 return result; 914 } 915 @Override 916 public boolean equals(Object obj) { 917 if (this == obj) 918 return true; 919 if (obj == null) 920 return false; 921 if (getClass() != obj.getClass()) 922 return false; 923 VirtualConsumerPair other = (VirtualConsumerPair) obj; 924 if (!getOuterType().equals(other.getOuterType())) 925 return false; 926 if (activeMQDestination == null) { 927 if (other.activeMQDestination != null) 928 return false; 929 } else if (!activeMQDestination.equals(other.activeMQDestination)) 930 return false; 931 if (virtualDestination == null) { 932 if (other.virtualDestination != null) 933 return false; 934 } else if (!virtualDestination.equals(other.virtualDestination)) 935 return false; 936 return true; 937 } 938 private AdvisoryBroker getOuterType() { 939 return AdvisoryBroker.this; 940 } 941 } 942}