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;
020
021import javax.jms.Destination;
022import javax.jms.JMSException;
023
024import org.apache.activemq.ActiveMQMessageTransformation;
025import org.apache.activemq.command.ActiveMQDestination;
026import org.apache.activemq.command.ActiveMQTopic;
027
028public final class AdvisorySupport {
029    public static final String ADVISORY_TOPIC_PREFIX = "ActiveMQ.Advisory.";
030    public static final ActiveMQTopic CONNECTION_ADVISORY_TOPIC = new ActiveMQTopic(ADVISORY_TOPIC_PREFIX
031            + "Connection");
032    public static final ActiveMQTopic QUEUE_ADVISORY_TOPIC = new ActiveMQTopic(ADVISORY_TOPIC_PREFIX + "Queue");
033    public static final ActiveMQTopic TOPIC_ADVISORY_TOPIC = new ActiveMQTopic(ADVISORY_TOPIC_PREFIX + "Topic");
034    public static final ActiveMQTopic TEMP_QUEUE_ADVISORY_TOPIC = new ActiveMQTopic(ADVISORY_TOPIC_PREFIX + "TempQueue");
035    public static final ActiveMQTopic TEMP_TOPIC_ADVISORY_TOPIC = new ActiveMQTopic(ADVISORY_TOPIC_PREFIX + "TempTopic");
036    public static final String PRODUCER_ADVISORY_TOPIC_PREFIX = ADVISORY_TOPIC_PREFIX + "Producer.";
037    public static final String QUEUE_PRODUCER_ADVISORY_TOPIC_PREFIX = PRODUCER_ADVISORY_TOPIC_PREFIX + "Queue.";
038    public static final String TOPIC_PRODUCER_ADVISORY_TOPIC_PREFIX = PRODUCER_ADVISORY_TOPIC_PREFIX + "Topic.";
039    public static final String CONSUMER_ADVISORY_TOPIC_PREFIX = ADVISORY_TOPIC_PREFIX + "Consumer.";
040    public static final String VIRTUAL_DESTINATION_CONSUMER_ADVISORY_TOPIC_PREFIX = ADVISORY_TOPIC_PREFIX + "VirtualDestination.Consumer.";
041    public static final String QUEUE_CONSUMER_ADVISORY_TOPIC_PREFIX = CONSUMER_ADVISORY_TOPIC_PREFIX + "Queue.";
042    public static final String TOPIC_CONSUMER_ADVISORY_TOPIC_PREFIX = CONSUMER_ADVISORY_TOPIC_PREFIX + "Topic.";
043    public static final String QUEUE_VIRTUAL_DESTINATION_CONSUMER_ADVISORY_TOPIC_PREFIX = VIRTUAL_DESTINATION_CONSUMER_ADVISORY_TOPIC_PREFIX + "Queue.";
044    public static final String TOPIC_VIRTUAL_DESTINATION_CONSUMER_ADVISORY_TOPIC_PREFIX = VIRTUAL_DESTINATION_CONSUMER_ADVISORY_TOPIC_PREFIX + "Topic.";
045    public static final String EXPIRED_TOPIC_MESSAGES_TOPIC_PREFIX = ADVISORY_TOPIC_PREFIX + "Expired.Topic.";
046    public static final String EXPIRED_QUEUE_MESSAGES_TOPIC_PREFIX = ADVISORY_TOPIC_PREFIX + "Expired.Queue.";
047    public static final String NO_TOPIC_CONSUMERS_TOPIC_PREFIX = ADVISORY_TOPIC_PREFIX + "NoConsumer.Topic.";
048    public static final String NO_QUEUE_CONSUMERS_TOPIC_PREFIX = ADVISORY_TOPIC_PREFIX + "NoConsumer.Queue.";
049    public static final String SLOW_CONSUMER_TOPIC_PREFIX = ADVISORY_TOPIC_PREFIX + "SlowConsumer.";
050    public static final String FAST_PRODUCER_TOPIC_PREFIX = ADVISORY_TOPIC_PREFIX + "FastProducer.";
051    public static final String MESSAGE_DISCAREDED_TOPIC_PREFIX = ADVISORY_TOPIC_PREFIX + "MessageDiscarded.";
052    public static final String FULL_TOPIC_PREFIX = ADVISORY_TOPIC_PREFIX + "FULL.";
053    public static final String MESSAGE_DELIVERED_TOPIC_PREFIX = ADVISORY_TOPIC_PREFIX + "MessageDelivered.";
054    public static final String MESSAGE_CONSUMED_TOPIC_PREFIX = ADVISORY_TOPIC_PREFIX + "MessageConsumed.";
055    public static final String MESSAGE_DLQ_TOPIC_PREFIX = ADVISORY_TOPIC_PREFIX + "MessageDLQd.";
056    public static final String MASTER_BROKER_TOPIC_PREFIX = ADVISORY_TOPIC_PREFIX + "MasterBroker";
057    public static final String NETWORK_BRIDGE_TOPIC_PREFIX = ADVISORY_TOPIC_PREFIX + "NetworkBridge";
058    public static final String NETWORK_BRIDGE_FORWARD_FAILURE_TOPIC_PREFIX = NETWORK_BRIDGE_TOPIC_PREFIX + ".ForwardFailure";
059    public static final String AGENT_TOPIC = "ActiveMQ.Agent";
060    public static final String ADIVSORY_MESSAGE_TYPE = "Advisory";
061    public static final String MSG_PROPERTY_ORIGIN_BROKER_ID = "originBrokerId";
062    public static final String MSG_PROPERTY_ORIGIN_BROKER_NAME = "originBrokerName";
063    public static final String MSG_PROPERTY_ORIGIN_BROKER_URL = "originBrokerURL";
064    public static final String MSG_PROPERTY_USAGE_NAME = "usageName";
065    public static final String MSG_PROPERTY_CONSUMER_ID = "consumerId";
066    public static final String MSG_PROPERTY_PRODUCER_ID = "producerId";
067    public static final String MSG_PROPERTY_MESSAGE_ID = "orignalMessageId";
068    public static final String MSG_PROPERTY_DESTINATION = "orignalDestination";
069    public static final String MSG_PROPERTY_CONSUMER_COUNT = "consumerCount";
070    public static final String MSG_PROPERTY_DISCARDED_COUNT = "discardedCount";
071
072    public static final ActiveMQTopic ALL_DESTINATIONS_COMPOSITE_ADVISORY_TOPIC = new ActiveMQTopic(
073            TOPIC_ADVISORY_TOPIC.getPhysicalName() + "," + QUEUE_ADVISORY_TOPIC.getPhysicalName() + "," +
074                    TEMP_QUEUE_ADVISORY_TOPIC.getPhysicalName() + "," + TEMP_TOPIC_ADVISORY_TOPIC.getPhysicalName());
075    public static final ActiveMQTopic TEMP_DESTINATION_COMPOSITE_ADVISORY_TOPIC = new ActiveMQTopic(
076            TEMP_QUEUE_ADVISORY_TOPIC.getPhysicalName() + "," + TEMP_TOPIC_ADVISORY_TOPIC.getPhysicalName());
077    private static final ActiveMQTopic AGENT_TOPIC_DESTINATION = new ActiveMQTopic(AGENT_TOPIC);
078
079    private AdvisorySupport() {
080    }
081
082    public static ActiveMQTopic getConnectionAdvisoryTopic() {
083        return CONNECTION_ADVISORY_TOPIC;
084    }
085
086    public static ActiveMQTopic[] getAllDestinationAdvisoryTopics(Destination destination) throws JMSException {
087        return getAllDestinationAdvisoryTopics(ActiveMQMessageTransformation.transformDestination(destination));
088    }
089
090    public static ActiveMQTopic[] getAllDestinationAdvisoryTopics(ActiveMQDestination destination) throws JMSException {
091        ArrayList<ActiveMQTopic> result = new ArrayList<ActiveMQTopic>();
092
093        result.add(getConsumerAdvisoryTopic(destination));
094        result.add(getProducerAdvisoryTopic(destination));
095        result.add(getExpiredMessageTopic(destination));
096        result.add(getNoConsumersAdvisoryTopic(destination));
097        result.add(getSlowConsumerAdvisoryTopic(destination));
098        result.add(getFastProducerAdvisoryTopic(destination));
099        result.add(getMessageDiscardedAdvisoryTopic(destination));
100        result.add(getMessageDeliveredAdvisoryTopic(destination));
101        result.add(getMessageConsumedAdvisoryTopic(destination));
102        result.add(getMessageDLQdAdvisoryTopic(destination));
103        result.add(getFullAdvisoryTopic(destination));
104
105        return result.toArray(new ActiveMQTopic[0]);
106    }
107
108    public static ActiveMQTopic getConsumerAdvisoryTopic(Destination destination) throws JMSException {
109        return getConsumerAdvisoryTopic(ActiveMQMessageTransformation.transformDestination(destination));
110    }
111
112    public static ActiveMQTopic getConsumerAdvisoryTopic(ActiveMQDestination destination) {
113        String prefix;
114        if (destination.isQueue()) {
115            prefix = QUEUE_CONSUMER_ADVISORY_TOPIC_PREFIX;
116        } else {
117            prefix = TOPIC_CONSUMER_ADVISORY_TOPIC_PREFIX;
118        }
119        return getAdvisoryTopic(destination, prefix, true);
120    }
121
122    public static ActiveMQTopic getVirtualDestinationConsumerAdvisoryTopic(ActiveMQDestination destination) {
123        String prefix;
124        if (destination.isQueue()) {
125            prefix = QUEUE_VIRTUAL_DESTINATION_CONSUMER_ADVISORY_TOPIC_PREFIX;
126        } else {
127            prefix = TOPIC_VIRTUAL_DESTINATION_CONSUMER_ADVISORY_TOPIC_PREFIX;
128        }
129        return getAdvisoryTopic(destination, prefix, true);
130    }
131
132    public static ActiveMQTopic getProducerAdvisoryTopic(Destination destination) throws JMSException {
133        return getProducerAdvisoryTopic(ActiveMQMessageTransformation.transformDestination(destination));
134    }
135
136    public static ActiveMQTopic getProducerAdvisoryTopic(ActiveMQDestination destination) {
137        String prefix;
138        if (destination.isQueue()) {
139            prefix = QUEUE_PRODUCER_ADVISORY_TOPIC_PREFIX;
140        } else {
141            prefix = TOPIC_PRODUCER_ADVISORY_TOPIC_PREFIX;
142        }
143        return getAdvisoryTopic(destination, prefix, false);
144    }
145
146    private static ActiveMQTopic getAdvisoryTopic(ActiveMQDestination destination, String prefix, boolean consumerTopics) {
147        return new ActiveMQTopic(prefix + destination.getPhysicalName().replaceAll(",", "&sbquo;"));
148    }
149
150    public static ActiveMQTopic getExpiredMessageTopic(Destination destination) throws JMSException {
151        return getExpiredMessageTopic(ActiveMQMessageTransformation.transformDestination(destination));
152    }
153
154    public static ActiveMQTopic getExpiredMessageTopic(ActiveMQDestination destination) {
155        if (destination.isQueue()) {
156            return getExpiredQueueMessageAdvisoryTopic(destination);
157        }
158        return getExpiredTopicMessageAdvisoryTopic(destination);
159    }
160
161    public static ActiveMQTopic getExpiredTopicMessageAdvisoryTopic(ActiveMQDestination destination) {
162        String name = EXPIRED_TOPIC_MESSAGES_TOPIC_PREFIX + destination.getPhysicalName();
163        return new ActiveMQTopic(name);
164    }
165
166    public static ActiveMQTopic getExpiredQueueMessageAdvisoryTopic(Destination destination) throws JMSException {
167        return getExpiredQueueMessageAdvisoryTopic(ActiveMQMessageTransformation.transformDestination(destination));
168    }
169
170    public static ActiveMQTopic getExpiredQueueMessageAdvisoryTopic(ActiveMQDestination destination) {
171        String name = EXPIRED_QUEUE_MESSAGES_TOPIC_PREFIX + destination.getPhysicalName();
172        return new ActiveMQTopic(name);
173    }
174
175    public static ActiveMQTopic getNoConsumersAdvisoryTopic(Destination destination) throws JMSException {
176        return getExpiredMessageTopic(ActiveMQMessageTransformation.transformDestination(destination));
177    }
178
179    public static ActiveMQTopic getNoConsumersAdvisoryTopic(ActiveMQDestination destination) {
180        if (destination.isQueue()) {
181            return getNoQueueConsumersAdvisoryTopic(destination);
182        }
183        return getNoTopicConsumersAdvisoryTopic(destination);
184    }
185
186    public static ActiveMQTopic getNoTopicConsumersAdvisoryTopic(Destination destination) throws JMSException {
187        return getNoTopicConsumersAdvisoryTopic(ActiveMQMessageTransformation.transformDestination(destination));
188    }
189
190    public static ActiveMQTopic getNoTopicConsumersAdvisoryTopic(ActiveMQDestination destination) {
191        String name = NO_TOPIC_CONSUMERS_TOPIC_PREFIX + destination.getPhysicalName();
192        return new ActiveMQTopic(name);
193    }
194
195    public static ActiveMQTopic getNoQueueConsumersAdvisoryTopic(Destination destination) throws JMSException {
196        return getNoQueueConsumersAdvisoryTopic(ActiveMQMessageTransformation.transformDestination(destination));
197    }
198
199    public static ActiveMQTopic getNoQueueConsumersAdvisoryTopic(ActiveMQDestination destination) {
200        String name = NO_QUEUE_CONSUMERS_TOPIC_PREFIX + destination.getPhysicalName();
201        return new ActiveMQTopic(name);
202    }
203
204    public static ActiveMQTopic getSlowConsumerAdvisoryTopic(Destination destination) throws JMSException {
205        return getSlowConsumerAdvisoryTopic(ActiveMQMessageTransformation.transformDestination(destination));
206    }
207
208    public static ActiveMQTopic getSlowConsumerAdvisoryTopic(ActiveMQDestination destination) {
209        String name = SLOW_CONSUMER_TOPIC_PREFIX + destination.getDestinationTypeAsString() + "."
210                + destination.getPhysicalName();
211        return new ActiveMQTopic(name);
212    }
213
214    public static ActiveMQTopic getFastProducerAdvisoryTopic(Destination destination) throws JMSException {
215        return getFastProducerAdvisoryTopic(ActiveMQMessageTransformation.transformDestination(destination));
216    }
217
218    public static ActiveMQTopic getFastProducerAdvisoryTopic(ActiveMQDestination destination) {
219        String name = FAST_PRODUCER_TOPIC_PREFIX + destination.getDestinationTypeAsString() + "."
220                + destination.getPhysicalName();
221        return new ActiveMQTopic(name);
222    }
223
224    public static ActiveMQTopic getMessageDiscardedAdvisoryTopic(Destination destination) throws JMSException {
225        return getMessageDiscardedAdvisoryTopic(ActiveMQMessageTransformation.transformDestination(destination));
226    }
227
228    public static ActiveMQTopic getMessageDiscardedAdvisoryTopic(ActiveMQDestination destination) {
229        String name = MESSAGE_DISCAREDED_TOPIC_PREFIX + destination.getDestinationTypeAsString() + "."
230                + destination.getPhysicalName();
231        return new ActiveMQTopic(name);
232    }
233
234    public static ActiveMQTopic getMessageDeliveredAdvisoryTopic(Destination destination) throws JMSException {
235        return getMessageDeliveredAdvisoryTopic(ActiveMQMessageTransformation.transformDestination(destination));
236    }
237
238    public static ActiveMQTopic getMessageDeliveredAdvisoryTopic(ActiveMQDestination destination) {
239        String name = MESSAGE_DELIVERED_TOPIC_PREFIX + destination.getDestinationTypeAsString() + "."
240                + destination.getPhysicalName();
241        return new ActiveMQTopic(name);
242    }
243
244    public static ActiveMQTopic getMessageConsumedAdvisoryTopic(Destination destination) throws JMSException {
245        return getMessageConsumedAdvisoryTopic(ActiveMQMessageTransformation.transformDestination(destination));
246    }
247
248    public static ActiveMQTopic getMessageConsumedAdvisoryTopic(ActiveMQDestination destination) {
249        String name = MESSAGE_CONSUMED_TOPIC_PREFIX + destination.getDestinationTypeAsString() + "."
250                + destination.getPhysicalName();
251        return new ActiveMQTopic(name);
252    }
253
254    public static ActiveMQTopic getMessageDLQdAdvisoryTopic(ActiveMQDestination destination) {
255        String name = MESSAGE_DLQ_TOPIC_PREFIX + destination.getDestinationTypeAsString() + "."
256                + destination.getPhysicalName();
257        return new ActiveMQTopic(name);
258    }
259
260    public static ActiveMQTopic getMasterBrokerAdvisoryTopic() {
261        return new ActiveMQTopic(MASTER_BROKER_TOPIC_PREFIX);
262    }
263
264    public static ActiveMQTopic getNetworkBridgeAdvisoryTopic() {
265        return new ActiveMQTopic(NETWORK_BRIDGE_TOPIC_PREFIX);
266    }
267
268    public static ActiveMQTopic getFullAdvisoryTopic(Destination destination) throws JMSException {
269        return getFullAdvisoryTopic(ActiveMQMessageTransformation.transformDestination(destination));
270    }
271
272    public static ActiveMQTopic getFullAdvisoryTopic(ActiveMQDestination destination) {
273        String name = FULL_TOPIC_PREFIX + destination.getDestinationTypeAsString() + "."
274                + destination.getPhysicalName();
275        return new ActiveMQTopic(name);
276    }
277
278    public static ActiveMQTopic getDestinationAdvisoryTopic(Destination destination) throws JMSException {
279        return getDestinationAdvisoryTopic(ActiveMQMessageTransformation.transformDestination(destination));
280    }
281
282    public static ActiveMQTopic getDestinationAdvisoryTopic(ActiveMQDestination destination) {
283        switch (destination.getDestinationType()) {
284            case ActiveMQDestination.QUEUE_TYPE:
285                return QUEUE_ADVISORY_TOPIC;
286            case ActiveMQDestination.TOPIC_TYPE:
287                return TOPIC_ADVISORY_TOPIC;
288            case ActiveMQDestination.TEMP_QUEUE_TYPE:
289                return TEMP_QUEUE_ADVISORY_TOPIC;
290            case ActiveMQDestination.TEMP_TOPIC_TYPE:
291                return TEMP_TOPIC_ADVISORY_TOPIC;
292            default:
293                throw new RuntimeException("Unknown destination type: " + destination.getDestinationType());
294        }
295    }
296
297    public static boolean isDestinationAdvisoryTopic(Destination destination) throws JMSException {
298        return isDestinationAdvisoryTopic(ActiveMQMessageTransformation.transformDestination(destination));
299    }
300
301    public static boolean isTempDestinationAdvisoryTopic(ActiveMQDestination destination) {
302        if (destination.isComposite()) {
303            ActiveMQDestination[] compositeDestinations = destination.getCompositeDestinations();
304            for (int i = 0; i < compositeDestinations.length; i++) {
305                if (!isTempDestinationAdvisoryTopic(compositeDestinations[i])) {
306                    return false;
307                }
308            }
309            return true;
310        } else {
311            return destination.equals(TEMP_QUEUE_ADVISORY_TOPIC) || destination.equals(TEMP_TOPIC_ADVISORY_TOPIC);
312        }
313    }
314
315    public static boolean isDestinationAdvisoryTopic(ActiveMQDestination destination) {
316        if (destination.isComposite()) {
317            ActiveMQDestination[] compositeDestinations = destination.getCompositeDestinations();
318            for (int i = 0; i < compositeDestinations.length; i++) {
319                if (isDestinationAdvisoryTopic(compositeDestinations[i])) {
320                    return true;
321                }
322            }
323            return false;
324        } else {
325            return destination.equals(TEMP_QUEUE_ADVISORY_TOPIC) || destination.equals(TEMP_TOPIC_ADVISORY_TOPIC)
326                    || destination.equals(QUEUE_ADVISORY_TOPIC) || destination.equals(TOPIC_ADVISORY_TOPIC);
327        }
328    }
329
330    public static boolean isAdvisoryTopic(Destination destination) throws JMSException {
331        return isAdvisoryTopic(ActiveMQMessageTransformation.transformDestination(destination));
332    }
333
334    public static boolean isAdvisoryTopic(ActiveMQDestination destination) {
335        if (destination != null) {
336            if (destination.isComposite()) {
337                ActiveMQDestination[] compositeDestinations = destination.getCompositeDestinations();
338                for (int i = 0; i < compositeDestinations.length; i++) {
339                    if (isAdvisoryTopic(compositeDestinations[i])) {
340                        return true;
341                    }
342                }
343                return false;
344            } else {
345                return destination.isTopic() && destination.getPhysicalName().startsWith(ADVISORY_TOPIC_PREFIX);
346            }
347        }
348        return false;
349    }
350
351    public static boolean isConnectionAdvisoryTopic(Destination destination) throws JMSException {
352        return isConnectionAdvisoryTopic(ActiveMQMessageTransformation.transformDestination(destination));
353    }
354
355    public static boolean isConnectionAdvisoryTopic(ActiveMQDestination destination) {
356        if (destination.isComposite()) {
357            ActiveMQDestination[] compositeDestinations = destination.getCompositeDestinations();
358            for (int i = 0; i < compositeDestinations.length; i++) {
359                if (isConnectionAdvisoryTopic(compositeDestinations[i])) {
360                    return true;
361                }
362            }
363            return false;
364        } else {
365            return destination.equals(CONNECTION_ADVISORY_TOPIC);
366        }
367    }
368
369    public static boolean isProducerAdvisoryTopic(Destination destination) throws JMSException {
370        return isProducerAdvisoryTopic(ActiveMQMessageTransformation.transformDestination(destination));
371    }
372
373    public static boolean isProducerAdvisoryTopic(ActiveMQDestination destination) {
374        if (destination.isComposite()) {
375            ActiveMQDestination[] compositeDestinations = destination.getCompositeDestinations();
376            for (int i = 0; i < compositeDestinations.length; i++) {
377                if (isProducerAdvisoryTopic(compositeDestinations[i])) {
378                    return true;
379                }
380            }
381            return false;
382        } else {
383            return destination.isTopic() && destination.getPhysicalName().startsWith(PRODUCER_ADVISORY_TOPIC_PREFIX);
384        }
385    }
386
387    public static boolean isConsumerAdvisoryTopic(Destination destination) throws JMSException {
388        return isConsumerAdvisoryTopic(ActiveMQMessageTransformation.transformDestination(destination));
389    }
390
391    public static boolean isConsumerAdvisoryTopic(ActiveMQDestination destination) {
392        if (destination.isComposite()) {
393            ActiveMQDestination[] compositeDestinations = destination.getCompositeDestinations();
394            for (int i = 0; i < compositeDestinations.length; i++) {
395                if (isConsumerAdvisoryTopic(compositeDestinations[i])) {
396                    return true;
397                }
398            }
399            return false;
400        } else {
401            return destination.isTopic() && destination.getPhysicalName().startsWith(CONSUMER_ADVISORY_TOPIC_PREFIX);
402        }
403    }
404
405    public static boolean isVirtualDestinationConsumerAdvisoryTopic(Destination destination) throws JMSException {
406        return isVirtualDestinationConsumerAdvisoryTopic(ActiveMQMessageTransformation.transformDestination(destination));
407    }
408
409    public static boolean isVirtualDestinationConsumerAdvisoryTopic(ActiveMQDestination destination) {
410        if (destination.isComposite()) {
411            ActiveMQDestination[] compositeDestinations = destination.getCompositeDestinations();
412            for (int i = 0; i < compositeDestinations.length; i++) {
413                if (isVirtualDestinationConsumerAdvisoryTopic(compositeDestinations[i])) {
414                    return true;
415                }
416            }
417            return false;
418        } else {
419            return destination.isTopic() && destination.getPhysicalName().startsWith(VIRTUAL_DESTINATION_CONSUMER_ADVISORY_TOPIC_PREFIX);
420        }
421    }
422
423    public static boolean isSlowConsumerAdvisoryTopic(Destination destination) throws JMSException {
424        return isSlowConsumerAdvisoryTopic(ActiveMQMessageTransformation.transformDestination(destination));
425    }
426
427    public static boolean isSlowConsumerAdvisoryTopic(ActiveMQDestination destination) {
428        if (destination.isComposite()) {
429            ActiveMQDestination[] compositeDestinations = destination.getCompositeDestinations();
430            for (int i = 0; i < compositeDestinations.length; i++) {
431                if (isSlowConsumerAdvisoryTopic(compositeDestinations[i])) {
432                    return true;
433                }
434            }
435            return false;
436        } else {
437            return destination.isTopic() && destination.getPhysicalName().startsWith(SLOW_CONSUMER_TOPIC_PREFIX);
438        }
439    }
440
441    public static boolean isFastProducerAdvisoryTopic(Destination destination) throws JMSException {
442        return isFastProducerAdvisoryTopic(ActiveMQMessageTransformation.transformDestination(destination));
443    }
444
445    public static boolean isFastProducerAdvisoryTopic(ActiveMQDestination destination) {
446        if (destination.isComposite()) {
447            ActiveMQDestination[] compositeDestinations = destination.getCompositeDestinations();
448            for (int i = 0; i < compositeDestinations.length; i++) {
449                if (isFastProducerAdvisoryTopic(compositeDestinations[i])) {
450                    return true;
451                }
452            }
453            return false;
454        } else {
455            return destination.isTopic() && destination.getPhysicalName().startsWith(FAST_PRODUCER_TOPIC_PREFIX);
456        }
457    }
458
459    public static boolean isMessageConsumedAdvisoryTopic(Destination destination) throws JMSException {
460        return isMessageConsumedAdvisoryTopic(ActiveMQMessageTransformation.transformDestination(destination));
461    }
462
463    public static boolean isMessageConsumedAdvisoryTopic(ActiveMQDestination destination) {
464        if (destination.isComposite()) {
465            ActiveMQDestination[] compositeDestinations = destination.getCompositeDestinations();
466            for (int i = 0; i < compositeDestinations.length; i++) {
467                if (isMessageConsumedAdvisoryTopic(compositeDestinations[i])) {
468                    return true;
469                }
470            }
471            return false;
472        } else {
473            return destination.isTopic() && destination.getPhysicalName().startsWith(MESSAGE_CONSUMED_TOPIC_PREFIX);
474        }
475    }
476
477    public static boolean isMasterBrokerAdvisoryTopic(Destination destination) throws JMSException {
478        return isMasterBrokerAdvisoryTopic(ActiveMQMessageTransformation.transformDestination(destination));
479    }
480
481    public static boolean isMasterBrokerAdvisoryTopic(ActiveMQDestination destination) {
482        if (destination.isComposite()) {
483            ActiveMQDestination[] compositeDestinations = destination.getCompositeDestinations();
484            for (int i = 0; i < compositeDestinations.length; i++) {
485                if (isMasterBrokerAdvisoryTopic(compositeDestinations[i])) {
486                    return true;
487                }
488            }
489            return false;
490        } else {
491            return destination.isTopic() && destination.getPhysicalName().startsWith(MASTER_BROKER_TOPIC_PREFIX);
492        }
493    }
494
495    public static boolean isMessageDeliveredAdvisoryTopic(Destination destination) throws JMSException {
496        return isMessageDeliveredAdvisoryTopic(ActiveMQMessageTransformation.transformDestination(destination));
497    }
498
499    public static boolean isMessageDeliveredAdvisoryTopic(ActiveMQDestination destination) {
500        if (destination.isComposite()) {
501            ActiveMQDestination[] compositeDestinations = destination.getCompositeDestinations();
502            for (int i = 0; i < compositeDestinations.length; i++) {
503                if (isMessageDeliveredAdvisoryTopic(compositeDestinations[i])) {
504                    return true;
505                }
506            }
507            return false;
508        } else {
509            return destination.isTopic() && destination.getPhysicalName().startsWith(MESSAGE_DELIVERED_TOPIC_PREFIX);
510        }
511    }
512
513    public static boolean isMessageDiscardedAdvisoryTopic(Destination destination) throws JMSException {
514        return isMessageDiscardedAdvisoryTopic(ActiveMQMessageTransformation.transformDestination(destination));
515    }
516
517    public static boolean isMessageDiscardedAdvisoryTopic(ActiveMQDestination destination) {
518        if (destination.isComposite()) {
519            ActiveMQDestination[] compositeDestinations = destination.getCompositeDestinations();
520            for (int i = 0; i < compositeDestinations.length; i++) {
521                if (isMessageDiscardedAdvisoryTopic(compositeDestinations[i])) {
522                    return true;
523                }
524            }
525            return false;
526        } else {
527            return destination.isTopic() && destination.getPhysicalName().startsWith(MESSAGE_DISCAREDED_TOPIC_PREFIX);
528        }
529    }
530
531    public static boolean isFullAdvisoryTopic(Destination destination) throws JMSException {
532        return isFullAdvisoryTopic(ActiveMQMessageTransformation.transformDestination(destination));
533    }
534
535    public static boolean isFullAdvisoryTopic(ActiveMQDestination destination) {
536        if (destination.isComposite()) {
537            ActiveMQDestination[] compositeDestinations = destination.getCompositeDestinations();
538            for (int i = 0; i < compositeDestinations.length; i++) {
539                if (isFullAdvisoryTopic(compositeDestinations[i])) {
540                    return true;
541                }
542            }
543            return false;
544        } else {
545            return destination.isTopic() && destination.getPhysicalName().startsWith(FULL_TOPIC_PREFIX);
546        }
547    }
548
549    public static boolean isNetworkBridgeAdvisoryTopic(Destination destination) throws JMSException {
550        return isNetworkBridgeAdvisoryTopic(ActiveMQMessageTransformation.transformDestination(destination));
551    }
552
553    public static boolean isNetworkBridgeAdvisoryTopic(ActiveMQDestination destination) {
554        if (destination.isComposite()) {
555            ActiveMQDestination[] compositeDestinations = destination.getCompositeDestinations();
556            for (int i = 0; i < compositeDestinations.length; i++) {
557                if (isNetworkBridgeAdvisoryTopic(compositeDestinations[i])) {
558                    return true;
559                }
560            }
561            return false;
562        } else {
563            return destination.isTopic() && destination.getPhysicalName().startsWith(NETWORK_BRIDGE_TOPIC_PREFIX);
564        }
565    }
566
567    /**
568     * Returns the agent topic which is used to send commands to the broker
569     */
570    public static Destination getAgentDestination() {
571        return AGENT_TOPIC_DESTINATION;
572    }
573
574    public static ActiveMQTopic getNetworkBridgeForwardFailureAdvisoryTopic() {
575        return new ActiveMQTopic(NETWORK_BRIDGE_FORWARD_FAILURE_TOPIC_PREFIX);
576    }
577}