From db4f602e8ce2fa85242e2737354252de882bc803 Mon Sep 17 00:00:00 2001 From: Robert Davies Date: Fri, 30 May 2008 11:36:47 +0000 Subject: [PATCH] Further tidying up for https://issues.apache.org/activemq/browse/AMQ-1704 git-svn-id: https://svn.apache.org/repos/asf/activemq/trunk@661661 13f79535-47bb-0310-9956-ffa450edef68 --- .../activemq/advisory/AdvisoryBroker.java | 44 +++++++++++-------- .../activemq/advisory/AdvisorySupport.java | 7 ++- 2 files changed, 32 insertions(+), 19 deletions(-) diff --git a/activemq-core/src/main/java/org/apache/activemq/advisory/AdvisoryBroker.java b/activemq-core/src/main/java/org/apache/activemq/advisory/AdvisoryBroker.java index 88ccfc4027..11478b3edd 100755 --- a/activemq-core/src/main/java/org/apache/activemq/advisory/AdvisoryBroker.java +++ b/activemq-core/src/main/java/org/apache/activemq/advisory/AdvisoryBroker.java @@ -253,7 +253,9 @@ public class AdvisoryBroker extends BrokerFilter { ActiveMQTopic topic = AdvisorySupport.getExpiredMessageTopic(messageReference.getMessage().getDestination()); Message payload = messageReference.getMessage().copy(); payload.clearBody(); - fireAdvisory(context, topic,payload); + ActiveMQMessage advisoryMessage = new ActiveMQMessage(); + advisoryMessage.setStringProperty(AdvisorySupport.MSG_PROPERTY_MESSAGE_ID, payload.getMessageId().toString()); + fireAdvisory(context, topic, payload, null, advisoryMessage); } } catch (Exception e) { LOG.warn("Failed to fire message expired advisory"); @@ -306,7 +308,9 @@ public class AdvisoryBroker extends BrokerFilter { super.slowConsumer(context, destination,subs); try { ActiveMQTopic topic = AdvisorySupport.getSlowConsumerAdvisoryTopic(destination.getActiveMQDestination()); - fireAdvisory(context, topic,subs.getConsumerInfo()); + ActiveMQMessage advisoryMessage = new ActiveMQMessage(); + advisoryMessage.setStringProperty(AdvisorySupport.MSG_PROPERTY_CONSUMER_ID, subs.getConsumerInfo().getConsumerId().toString()); + fireAdvisory(context, topic, subs.getConsumerInfo(), null, advisoryMessage); } catch (Exception e) { LOG.warn("Failed to fire message slow consumer advisory"); } @@ -316,7 +320,9 @@ public class AdvisoryBroker extends BrokerFilter { super.fastProducer(context, producerInfo); try { ActiveMQTopic topic = AdvisorySupport.getFastProducerAdvisoryTopic(producerInfo.getDestination()); - fireAdvisory(context, topic,producerInfo); + ActiveMQMessage advisoryMessage = new ActiveMQMessage(); + advisoryMessage.setStringProperty(AdvisorySupport.MSG_PROPERTY_PRODUCER_ID, producerInfo.getProducerId().toString()); + fireAdvisory(context, topic, producerInfo, null, advisoryMessage); } catch (Exception e) { LOG.warn("Failed to fire message fast producer advisory"); } @@ -327,8 +333,8 @@ public class AdvisoryBroker extends BrokerFilter { try { ActiveMQTopic topic = AdvisorySupport.getFullAdvisoryTopic(destination.getActiveMQDestination()); ActiveMQMessage advisoryMessage = new ActiveMQMessage(); - advisoryMessage.setStringProperty("usageName", usage.getName()); - fireAdvisory(context, topic,advisoryMessage); + advisoryMessage.setStringProperty(AdvisorySupport.MSG_PROPERTY_USAGE_NAME, usage.getName()); + fireAdvisory(context, topic,null,null,advisoryMessage); } catch (Exception e) { LOG.warn("Failed to fire message is full advisory"); } @@ -338,18 +344,10 @@ public class AdvisoryBroker extends BrokerFilter { super.nowMasterBroker(); try { ActiveMQTopic topic = AdvisorySupport.getMasterBrokerAdvisoryTopic(); - ActiveMQMessage advisoryMessage = new ActiveMQMessage(); - advisoryMessage.setStringProperty("brokerName", getBrokerName()); - String[] uris = getBrokerService().getTransportConnectorURIs(); - String uri = getBrokerService().getVmConnectorURI().toString(); - if (uris != null && uris.length > 0) { - uri = uris[0]; - } - advisoryMessage.setStringProperty("brokerURL", getBrokerName()); - advisoryMessage.setStringProperty("brokerURI", uri); + ActiveMQMessage advisoryMessage = new ActiveMQMessage(); ConnectionContext context = new ConnectionContext(); context.setBroker(getBrokerService().getBroker()); - fireAdvisory(context, topic,advisoryMessage); + fireAdvisory(context, topic,null,null,advisoryMessage); } catch (Exception e) { LOG.warn("Failed to fire message master broker advisory"); } @@ -361,9 +359,6 @@ public class AdvisoryBroker extends BrokerFilter { protected void fireAdvisory(ConnectionContext context, ActiveMQTopic topic, Command command, ConsumerId targetConsumerId) throws Exception { ActiveMQMessage advisoryMessage = new ActiveMQMessage(); - advisoryMessage.setStringProperty(AdvisorySupport.MSG_PROPERTY_ORIGIN_BROKER_NAME, getBrokerName()); - String id = getBrokerId() != null ? getBrokerId().getValue() : "NOT_SET"; - advisoryMessage.setStringProperty(AdvisorySupport.MSG_PROPERTY_ORIGIN_BROKER_ID, id); fireAdvisory(context, topic, command, targetConsumerId, advisoryMessage); } @@ -406,6 +401,19 @@ public class AdvisoryBroker extends BrokerFilter { protected void fireAdvisory(ConnectionContext context, ActiveMQTopic topic, Command command, ConsumerId targetConsumerId, ActiveMQMessage advisoryMessage) throws Exception { if (getBrokerService().isStarted()) { + //set properties + advisoryMessage.setStringProperty(AdvisorySupport.MSG_PROPERTY_ORIGIN_BROKER_NAME, getBrokerName()); + String id = getBrokerId() != null ? getBrokerId().getValue() : "NOT_SET"; + advisoryMessage.setStringProperty(AdvisorySupport.MSG_PROPERTY_ORIGIN_BROKER_ID, id); + + String[] uris = getBrokerService().getTransportConnectorURIs(); + String url = getBrokerService().getVmConnectorURI().toString(); + if (uris != null && uris.length > 0) { + url = uris[0]; + } + advisoryMessage.setStringProperty(AdvisorySupport.MSG_PROPERTY_ORIGIN_BROKER_URL, url); + + //set the data structure advisoryMessage.setDataStructure(command); advisoryMessage.setPersistent(false); advisoryMessage.setType(AdvisorySupport.ADIVSORY_MESSAGE_TYPE); diff --git a/activemq-core/src/main/java/org/apache/activemq/advisory/AdvisorySupport.java b/activemq-core/src/main/java/org/apache/activemq/advisory/AdvisorySupport.java index 1c05f174cc..619ea4ca29 100755 --- a/activemq-core/src/main/java/org/apache/activemq/advisory/AdvisorySupport.java +++ b/activemq-core/src/main/java/org/apache/activemq/advisory/AdvisorySupport.java @@ -40,7 +40,7 @@ public final class AdvisorySupport { public static final String NO_TOPIC_CONSUMERS_TOPIC_PREFIX = ADVISORY_TOPIC_PREFIX + "NoConsumer.Topic."; public static final String NO_QUEUE_CONSUMERS_TOPIC_PREFIX = ADVISORY_TOPIC_PREFIX + "NoConsumer.Queue."; public static final String SLOW_CONSUMER_TOPIC_PREFIX = ADVISORY_TOPIC_PREFIX + "SlowConsumer."; - public static final String FAST_PRODUCER_TOPIC_PREFIX = ADVISORY_TOPIC_PREFIX + "FastConsumer."; + public static final String FAST_PRODUCER_TOPIC_PREFIX = ADVISORY_TOPIC_PREFIX + "FastPorducer."; public static final String MESSAGE_DISCAREDED_TOPIC_PREFIX = ADVISORY_TOPIC_PREFIX + "MessageDiscarded."; public static final String FULL_TOPIC_PREFIX = ADVISORY_TOPIC_PREFIX + "FULL."; public static final String MESSAGE_DELIVERED_TOPIC_PREFIX = ADVISORY_TOPIC_PREFIX + "MessageDelivered."; @@ -50,6 +50,11 @@ public final class AdvisorySupport { public static final String ADIVSORY_MESSAGE_TYPE = "Advisory"; public static final String MSG_PROPERTY_ORIGIN_BROKER_ID="originBrokerId"; public static final String MSG_PROPERTY_ORIGIN_BROKER_NAME="originBrokerName"; + public static final String MSG_PROPERTY_ORIGIN_BROKER_URL="originBrokerURL"; + public static final String MSG_PROPERTY_USAGE_NAME="usageName"; + public static final String MSG_PROPERTY_CONSUMER_ID="consumerId"; + public static final String MSG_PROPERTY_PRODUCER_ID="producerId"; + public static final String MSG_PROPERTY_MESSAGE_ID="orignalMessageId"; public static final ActiveMQTopic TEMP_DESTINATION_COMPOSITE_ADVISORY_TOPIC = new ActiveMQTopic(TEMP_QUEUE_ADVISORY_TOPIC + "," + TEMP_TOPIC_ADVISORY_TOPIC); private static final ActiveMQTopic AGENT_TOPIC_DESTINATION = new ActiveMQTopic(AGENT_TOPIC);