activemq-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From rajdav...@apache.org
Subject svn commit: r661661 - in /activemq/trunk/activemq-core/src/main/java/org/apache/activemq/advisory: AdvisoryBroker.java AdvisorySupport.java
Date Fri, 30 May 2008 11:36:48 GMT
Author: rajdavies
Date: Fri May 30 04:36:47 2008
New Revision: 661661

URL: http://svn.apache.org/viewvc?rev=661661&view=rev
Log:
Further tidying up for https://issues.apache.org/activemq/browse/AMQ-1704

Modified:
    activemq/trunk/activemq-core/src/main/java/org/apache/activemq/advisory/AdvisoryBroker.java
    activemq/trunk/activemq-core/src/main/java/org/apache/activemq/advisory/AdvisorySupport.java

Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/advisory/AdvisoryBroker.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/advisory/AdvisoryBroker.java?rev=661661&r1=661660&r2=661661&view=diff
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/advisory/AdvisoryBroker.java
(original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/advisory/AdvisoryBroker.java
Fri May 30 04:36:47 2008
@@ -253,7 +253,9 @@
                 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 @@
         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 @@
         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 @@
         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 @@
         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 @@
 
     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 @@
 
     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);

Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/advisory/AdvisorySupport.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/advisory/AdvisorySupport.java?rev=661661&r1=661660&r2=661661&view=diff
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/advisory/AdvisorySupport.java
(original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/advisory/AdvisorySupport.java
Fri May 30 04:36:47 2008
@@ -40,7 +40,7 @@
     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 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);
 



Mime
View raw message