Apache
/
activemq
mirror of https://github.com/apache/activemq.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

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
This commit is contained in:
Robert Davies 2008-05-30 11:36:47 +00:00
parent 6cc18cac5a
commit db4f602e8c
2 changed files with 32 additions and 19 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

44
activemq-core/src/main/java/org/apache/activemq/advisory/AdvisoryBroker.java
View File

@ -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);

7
activemq-core/src/main/java/org/apache/activemq/advisory/AdvisorySupport.java
View File

@ -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);