You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@activemq.apache.org by ra...@apache.org on 2008/05/30 13:36:48 UTC
svn commit: r661661 - in
/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/advisory:
AdvisoryBroker.java AdvisorySupport.java
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);