Source code: org/activemq/message/ActiveMQDestination.java
1 /**
2 *
3 * Copyright 2004 Protique Ltd
4 *
5 * Licensed under the Apache License, Version 2.0 (the "License");
6 * you may not use this file except in compliance with the License.
7 * You may obtain a copy of the License at
8 *
9 * http://www.apache.org/licenses/LICENSE-2.0
10 *
11 * Unless required by applicable law or agreed to in writing, software
12 * distributed under the License is distributed on an "AS IS" BASIS,
13 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14 * See the License for the specific language governing permissions and
15 * limitations under the License.
16 *
17 **/
18
19 package org.activemq.message;
20
21 import java.io.DataInput;
22 import java.io.DataOutput;
23 import java.io.IOException;
24 import java.io.Serializable;
25 import java.util.ArrayList;
26 import java.util.List;
27 import java.util.Properties;
28 import java.util.StringTokenizer;
29 import javax.jms.Destination;
30 import javax.jms.JMSException;
31 import javax.jms.Queue;
32 import javax.jms.TemporaryQueue;
33 import javax.jms.TemporaryTopic;
34 import javax.jms.Topic;
35 import org.activemq.ActiveMQSession;
36 import org.activemq.filter.DestinationFilter;
37 import org.activemq.filter.DestinationPath;
38 import org.activemq.jndi.JNDIBaseStorable;
39 import org.activemq.management.JMSDestinationStats;
40
41 /**
42 * A <CODE>Destination</CODE> object encapsulates a provider-specific
43 * address.
44 * The JMS API does not define a standard address syntax. Although a standard
45 * address syntax was considered, it was decided that the differences in
46 * address semantics between existing message-oriented middleware (MOM)
47 * products were too wide to bridge with a single syntax.
48 * <p/>
49 * <P>Since <CODE>Destination</CODE> is an administered object, it may
50 * contain
51 * provider-specific configuration information in addition to its address.
52 * <p/>
53 * <P>The JMS API also supports a client's use of provider-specific address
54 * names.
55 * <p/>
56 * <P><CODE>Destination</CODE> objects support concurrent use.
57 * <p/>
58 * <P>A <CODE>Destination</CODE> object is a JMS administered object.
59 * <p/>
60 * <P>JMS administered objects are objects containing configuration
61 * information that are created by an administrator and later used by
62 * JMS clients. They make it practical to administer the JMS API in the
63 * enterprise.
64 * <p/>
65 * <P>Although the interfaces for administered objects do not explicitly
66 * depend on the Java Naming and Directory Interface (JNDI) API, the JMS API
67 * establishes the convention that JMS clients find administered objects by
68 * looking them up in a JNDI namespace.
69 * <p/>
70 * <P>An administrator can place an administered object anywhere in a
71 * namespace. The JMS API does not define a naming policy.
72 * <p/>
73 * <P>It is expected that JMS providers will provide the tools an
74 * administrator needs to create and configure administered objects in a
75 * JNDI namespace. JMS provider implementations of administered objects
76 * should implement the <CODE>javax.naming.Referenceable</CODE> and
77 * <CODE>java.io.Serializable</CODE> interfaces so that they can be stored in
78 * all JNDI naming contexts. In addition, it is recommended that these
79 * implementations follow the JavaBeans<SUP><FONT SIZE="-2">TM</FONT></SUP>
80 * design patterns.
81 * <p/>
82 * <P>This strategy provides several benefits:
83 * <p/>
84 * <UL>
85 * <LI>It hides provider-specific details from JMS clients.
86 * <LI>It abstracts JMS administrative information into objects in the Java
87 * programming language ("Java objects")
88 * that are easily organized and administered from a common
89 * management console.
90 * <LI>Since there will be JNDI providers for all popular naming
91 * services, JMS providers can deliver one implementation
92 * of administered objects that will run everywhere.
93 * </UL>
94 * <p/>
95 * <P>An administered object should not hold on to any remote resources.
96 * Its lookup should not use remote resources other than those used by the
97 * JNDI API itself.
98 * <p/>
99 * <P>Clients should think of administered objects as local Java objects.
100 * Looking them up should not have any hidden side effects or use surprising
101 * amounts of local resources.
102 */
103
104 public abstract class ActiveMQDestination extends JNDIBaseStorable implements Destination, Comparable, Serializable {
105
106 static final long serialVersionUID = -3300456112096957638L;
107
108 /**
109 * Topic Destination object
110 */
111 public static final int ACTIVEMQ_TOPIC = 1;
112 /**
113 * Temporary Topic Destination object
114 */
115 public static final int ACTIVEMQ_TEMPORARY_TOPIC = 2;
116
117 /**
118 * Queue Destination object
119 */
120 public static final int ACTIVEMQ_QUEUE = 3;
121 /**
122 * Temporary Queue Destination object
123 */
124 public static final int ACTIVEMQ_TEMPORARY_QUEUE = 4;
125
126 /**
127 * prefix for Advisory message destinations
128 */
129 public static final String ADVISORY_PREFIX = "ActiveMQ.Advisory.";
130
131 /**
132 * prefix for connection advisory destinations
133 */
134 public static final String CONNECTION_ADVISORY_PREFIX = ADVISORY_PREFIX + "Connections.";
135
136 /**
137 * prefix for consumer advisory destinations
138 * @deprecated Use {@see #getDestinationForConsumerAdvisory()} instead.
139 */
140 public static final String CONSUMER_ADVISORY_PREFIX = ADVISORY_PREFIX + "Consumers.";
141
142 /**
143 * prefix for producer advisory destinations
144 * @deprecated Use {@see #getDestinationForProducerAdvisory()} instead.
145 */
146 public static final String PRODUCER_ADVISORY_PREFIX = ADVISORY_PREFIX + "Producers.";
147
148 /**
149 * prefix for connection advisory destinations
150 * @deprecated Use {@see #getDestinationForTempAdvisory()} instead.
151 */
152 public static final String TEMP_DESTINATION_ADVISORY_PREFIX = ADVISORY_PREFIX + "TempDestinations.";
153
154 /**
155 * The default target for ordered destinations
156 */
157 public static final String DEFAULT_ORDERED_TARGET = "coordinator";
158
159 private static final int NULL_DESTINATION = 10;
160
161 private static final String TEMP_PREFIX = "{TD{";
162 private static final String TEMP_POSTFIX = "}TD}";
163 private static final String COMPOSITE_SEPARATOR = ",";
164
165 private static final String QUEUE_PREFIX = "queue://";
166 private static final String TOPIC_PREFIX = "topic://";
167
168 private String physicalName = "";
169
170 // Cached transient data
171 private transient DestinationFilter filter;
172 private transient JMSDestinationStats stats;
173 private transient String[] paths;
174 // Used track consumers of temporary topics.
175 private transient int consumerCounter;
176 private transient boolean deleted;
177 //additional for defining 'exotic' behaviour
178 private boolean exclusive;
179 private boolean ordered;
180 private boolean advisory;
181 private boolean wildcard;
182 private boolean composite;
183 private String orderedTarget = DEFAULT_ORDERED_TARGET;
184 //used client-side only
185 private transient ActiveMQSession sessionCreatedBy;
186
187 /**
188 * @return Returns the orginatingSession.
189 */
190 public ActiveMQSession getSessionCreatedBy() {
191 return sessionCreatedBy;
192 }
193 /**
194 * @param orginatingSession The orginatingSession to set.
195 */
196 public void setSessionCreatedBy(ActiveMQSession orginatingSession) {
197 this.sessionCreatedBy = orginatingSession;
198 }
199 /**
200 * The Default Constructor
201 */
202 protected ActiveMQDestination() {
203 }
204
205 /**
206 * Construct the ActiveMQDestination with a defined physical name;
207 *
208 * @param name
209 */
210
211 protected ActiveMQDestination(String name) {
212 setPhysicalName(name);
213 }
214
215 /**
216 * @return Returns the advisory.
217 */
218 public final boolean isAdvisory() {
219 return advisory;
220 }
221 /**
222 * @param advisory The advisory to set.
223 */
224 public final void setAdvisory(boolean advisory) {
225 this.advisory = advisory;
226 }
227
228 /**
229 * @return true if this is a destination for Consumer advisories
230 */
231 public boolean isConsumerAdvisory(){
232 return isAdvisory() && physicalName.startsWith(ActiveMQDestination.CONSUMER_ADVISORY_PREFIX);
233 }
234
235 /**
236 * @return true if this is a destination for Producer advisories
237 */
238 public boolean isProducerAdvisory(){
239 return isAdvisory() && physicalName.startsWith(ActiveMQDestination.PRODUCER_ADVISORY_PREFIX);
240 }
241
242 /**
243 * @return true if this is a destination for Connection advisories
244 */
245 public boolean isConnectionAdvisory(){
246 return isAdvisory() && physicalName.startsWith(ActiveMQDestination.CONNECTION_ADVISORY_PREFIX);
247 }
248
249 /**
250 * @return true if this a destination for Tempoary Destination advisories
251 */
252 public final boolean isTempDestinationAdvisory(){
253 return advisory && physicalName.startsWith(ActiveMQDestination.TEMP_DESTINATION_ADVISORY_PREFIX);
254 }
255
256 /**
257 * @return Returns the exclusive.
258 */
259 public final boolean isExclusive() {
260 return exclusive;
261 }
262 /**
263 * @param exclusive The exclusive to set.
264 */
265 public final void setExclusive(boolean exclusive) {
266 this.exclusive = exclusive;
267 }
268 /**
269 * @return Returns the ordered.
270 */
271 public final boolean isOrdered() {
272 return ordered;
273 }
274 /**
275 * @param ordered The ordered to set.
276 */
277 public final void setOrdered(boolean ordered) {
278 this.ordered = ordered;
279 }
280 /**
281 * @return Returns the orderedTarget.
282 */
283 public String getOrderedTarget() {
284 return orderedTarget;
285 }
286 /**
287 * @param orderedTarget The orderedTarget to set.
288 */
289 public void setOrderedTarget(String orderedTarget) {
290 this.orderedTarget = orderedTarget;
291 }
292 /**
293 * A helper method to return a descriptive string for the topic or queue
294 * @param destination
295 *
296 * @return a descriptive string for this queue or topic
297 */
298 public static String inspect(Destination destination) {
299 if (destination instanceof Topic) {
300 return "Topic(" + destination.toString() + ")";
301 }
302 else {
303 return "Queue(" + destination.toString() + ")";
304 }
305 }
306
307 /**
308 * @param destination
309 * @return @throws JMSException
310 * @throws javax.jms.JMSException
311 */
312 public static ActiveMQDestination transformDestination(Destination destination) throws JMSException {
313 ActiveMQDestination result = null;
314 if (destination != null) {
315 if (destination instanceof ActiveMQDestination) {
316 result = (ActiveMQDestination) destination;
317 }
318 else {
319 if (destination instanceof TemporaryQueue) {
320 result = new ActiveMQTemporaryQueue(((Queue) destination).getQueueName());
321 }
322 else if (destination instanceof TemporaryTopic) {
323 result = new ActiveMQTemporaryTopic(((Topic) destination).getTopicName());
324 }
325 else if (destination instanceof Queue) {
326 result = new ActiveMQTemporaryQueue(((Queue) destination).getQueueName());
327 }
328 else if (destination instanceof Topic) {
329 result = new ActiveMQTemporaryTopic(((Topic) destination).getTopicName());
330 }
331 }
332 }
333 return result;
334 }
335
336 /**
337 * Write an ActiveMQDestination to a Stream
338 *
339 * @param destination
340 * @param dataOut
341 * @throws IOException
342 */
343
344 public static void writeToStream(ActiveMQDestination destination, DataOutput dataOut) throws IOException {
345 if (destination != null) {
346 dataOut.writeByte(destination.getDestinationType());
347 String physicalName = destination.getPhysicalName();
348 boolean writeOrderedTarget = destination.orderedTarget != null && !destination.orderedTarget.equals(DEFAULT_ORDERED_TARGET);
349 byte byte1 = 0;
350 if (physicalName != null && physicalName.length() > 0) byte1 |= 1;
351 if (destination.ordered) byte1 |= 2;
352 if(destination.exclusive) byte1 |= 4;
353 if (writeOrderedTarget) byte1 |= 8;
354 if (destination.advisory) byte1 |= 16;
355 if (destination.deleted) byte1 |= 32;
356 if (destination.composite) byte1 |= 64;
357 if (destination.wildcard) byte1 |= 128;
358 dataOut.writeByte(byte1);
359 if (physicalName != null && physicalName.length() > 0){
360 dataOut.writeUTF(physicalName);
361 }
362 if (writeOrderedTarget){
363 dataOut.writeUTF(destination.orderedTarget);
364 }
365 }
366 else {
367 dataOut.write(NULL_DESTINATION);
368 }
369 }
370
371 /**
372 * Read an ActiveMQDestination from a Stream
373 *
374 * @param dataIn
375 * @return the ActiveMQDestination
376 * @throws IOException
377 */
378
379 public static ActiveMQDestination readFromStream(DataInput dataIn) throws IOException {
380
381 int type = dataIn.readByte();
382 if (type == NULL_DESTINATION) {
383 return null;
384 }
385 ActiveMQDestination result = null;
386 if (type == ACTIVEMQ_TOPIC) {
387 result = new ActiveMQTopic();
388 }
389 else if (type == ACTIVEMQ_TEMPORARY_TOPIC) {
390 result = new ActiveMQTemporaryTopic();
391 }
392 else if (type == ACTIVEMQ_QUEUE) {
393 result = new ActiveMQQueue();
394 }
395 else {
396 result = new ActiveMQTemporaryQueue();
397 }
398 byte byte1 = dataIn.readByte();
399 if ((byte1 & 1) == 1){
400 result.physicalName = dataIn.readUTF();
401 }
402 result.setOrdered((byte1 & 2) == 2);
403 result.setExclusive((byte1 & 4) == 4);
404 if ((byte1 & 8) == 8){
405 result.setOrderedTarget(dataIn.readUTF());
406 }
407 result.advisory = (byte1 & 16)==16;
408 result.deleted = (byte1 & 32)==32;
409 result.composite = (byte1 & 64)==64;
410 result.wildcard = (byte1 & 128)==128;
411 return result;
412 }
413
414 /**
415 * Create a Destination
416 * @param type
417 * @param pyhsicalName
418 * @return
419 */
420 public static ActiveMQDestination createDestination(int type,String pyhsicalName){
421 ActiveMQDestination result = null;
422 if (type == ACTIVEMQ_TOPIC) {
423 result = new ActiveMQTopic(pyhsicalName);
424 }
425 else if (type == ACTIVEMQ_TEMPORARY_TOPIC) {
426 result = new ActiveMQTemporaryTopic(pyhsicalName);
427 }
428 else if (type == ACTIVEMQ_QUEUE) {
429 result = new ActiveMQQueue(pyhsicalName);
430 }
431 else {
432 result = new ActiveMQTemporaryQueue(pyhsicalName);
433 }
434 return result;
435 }
436
437 /**
438 * Create a temporary name from the clientId
439 *
440 * @param clientId
441 * @return
442 */
443 public static String createTemporaryName(String clientId) {
444 return TEMP_PREFIX + clientId + TEMP_POSTFIX;
445 }
446
447 /**
448 * From a temporary destination find the clientId of the Connection that created it
449 *
450 * @param destination
451 * @return the clientId or null if not a temporary destination
452 */
453 public static String getClientId(ActiveMQDestination destination) {
454 String answer = null;
455 if (destination != null && destination.isTemporary()) {
456 String name = destination.getPhysicalName();
457 int start = name.indexOf(TEMP_PREFIX);
458 if (start >= 0) {
459 start += TEMP_PREFIX.length();
460 int stop = name.lastIndexOf(TEMP_POSTFIX);
461 if (stop > start && stop < name.length()) {
462 answer = name.substring(start, stop);
463 }
464 }
465 }
466 return answer;
467 }
468
469
470 /**
471 * @param o object to compare
472 * @return 1 if this > o else 0 if they are equal or -1 if this < o
473 */
474 public int compareTo(Object o) {
475 if (o instanceof ActiveMQDestination) {
476 return compareTo((ActiveMQDestination) o);
477 }
478 return -1;
479 }
480
481 /**
482 * Lets sort by name first then lets sort topics greater than queues
483 *
484 * @param that another destination to compare against
485 * @return 1 if this > that else 0 if they are equal or -1 if this < that
486 */
487 public int compareTo(ActiveMQDestination that) {
488 int answer = 0;
489 if (physicalName != that.physicalName) {
490 if (physicalName == null) {
491 return -1;
492 }
493 else if (that.physicalName == null) {
494 return 1;
495 }
496 answer = physicalName.compareTo(that.physicalName);
497 }
498 if (answer == 0) {
499 if (isTopic()) {
500 if (that.isQueue()) {
501 return 1;
502 }
503 }
504 else {
505 if (that.isTopic()) {
506 return -1;
507 }
508 }
509 }
510 return answer;
511 }
512
513
514 /**
515 * @return Returns the Destination type
516 */
517
518 public abstract int getDestinationType();
519
520
521 /**
522 * @return Returns the physicalName.
523 */
524 public String getPhysicalName() {
525 return this.physicalName;
526 }
527
528 /**
529 * @param name The physicalName to set.
530 */
531 public void setPhysicalName(String name) {
532 this.physicalName = name;
533 this.advisory = name != null && name.startsWith(ADVISORY_PREFIX);
534 this.composite = name != null && name.indexOf(COMPOSITE_SEPARATOR) > 0;
535 this.wildcard = name != null
536 && (name.indexOf(DestinationFilter.ANY_CHILD) >= 0 || name.indexOf(DestinationFilter.ANY_DESCENDENT) >= 0);
537 }
538
539 /**
540 * Returns true if a temporary Destination
541 *
542 * @return true/false
543 */
544
545 public boolean isTemporary() {
546 return false;
547 }
548
549 /**
550 * Returns true if a Topic Destination
551 *
552 * @return true/false
553 */
554
555 public boolean isTopic() {
556 return true;
557 }
558
559 /**
560 * Returns true if a Queue Destination
561 *
562 * @return true/false
563 */
564 public boolean isQueue() {
565 return false;
566 }
567
568 /**
569 * Returns true if this destination represents a collection of
570 * destinations; allowing a set of destinations to be published to or subscribed
571 * from in one JMS operation.
572 * <p/>
573 * If this destination is a composite then you can call {@link #getChildDestinations()}
574 * to return the list of child destinations.
575 *
576 * @return true if this destination represents a collection of child destinations.
577 */
578 public final boolean isComposite() {
579 return composite;
580 }
581
582 /**
583 * Returns a list of child destinations if this destination represents a composite
584 * destination.
585 *
586 * @return
587 */
588 public List getChildDestinations() {
589 List answer = new ArrayList();
590 StringTokenizer iter = new StringTokenizer(physicalName, COMPOSITE_SEPARATOR);
591 while (iter.hasMoreTokens()) {
592 String name = iter.nextToken();
593 Destination child = null;
594 if (name.startsWith(QUEUE_PREFIX)) {
595 child = new ActiveMQQueue(name.substring(QUEUE_PREFIX.length()));
596 }
597 else if (name.startsWith(TOPIC_PREFIX)) {
598 child = new ActiveMQTopic(name.substring(TOPIC_PREFIX.length()));
599 }
600 else {
601 child = createDestination(name);
602 }
603 answer.add(child);
604 }
605 if (answer.size() == 1) {
606 // lets put ourselves inside the collection
607 // as we are not really a composite destination
608 answer.set(0, this);
609 }
610 return answer;
611 }
612
613 public void setChildDestinations(ActiveMQDestination[] children) {
614 if( children==null )
615 throw new IllegalArgumentException("children array cannot be null.");
616 if( children.length == 0 )
617 throw new IllegalArgumentException("children array size must be 1 or greater.");
618
619 StringBuffer rc = new StringBuffer();
620 for (int i = 0; i < children.length; i++) {
621 if(i!=0)
622 rc.append(COMPOSITE_SEPARATOR);
623 if( children[i].isTopic() ) {
624 rc.append(TOPIC_PREFIX);
625 } else if( children[i].isQueue() ) {
626 rc.append(QUEUE_PREFIX);
627 }
628 rc.append(children[i].getPhysicalName());
629 }
630
631 setPhysicalName(rc.toString());
632 }
633
634 /**
635 * @return string representation of this instance
636 */
637
638 public String toString() {
639 return this.physicalName;
640 }
641
642 /**
643 * @return hashCode for this instance
644 */
645
646 public int hashCode() {
647 int answer = 0xcafebabe;
648
649 if (this.physicalName != null) {
650 answer = physicalName.hashCode();
651 }
652 if (isTopic()) {
653 answer ^= 0xfabfab;
654 }
655 return answer;
656 }
657
658 /**
659 * if the object passed in is equivalent, return true
660 *
661 * @param obj the object to compare
662 * @return true if this instance and obj are equivalent
663 */
664
665 public boolean equals(Object obj) {
666 boolean result = this == obj;
667 if (!result && obj != null && obj instanceof ActiveMQDestination) {
668 ActiveMQDestination other = (ActiveMQDestination) obj;
669 result = this.getDestinationType() == other.getDestinationType() &&
670 this.physicalName.equals(other.physicalName);
671 }
672 return result;
673 }
674
675
676 /**
677 * @return true if the destination matches multiple possible destinations
678 */
679 public boolean isWildcard() {
680 return wildcard;
681 }
682
683 /**
684 * @param destination
685 * @return true if the given destination matches this destination; including wildcards
686 */
687 public boolean matches(ActiveMQDestination destination) {
688 if (isWildcard()) {
689 return getDestinationFilter().matches(destination);
690 }
691 else {
692 return equals(destination);
693 }
694 }
695
696
697 /**
698 * @return the DestinationFilter
699 */
700 public DestinationFilter getDestinationFilter() {
701 if (filter == null) {
702 filter = DestinationFilter.parseFilter(this);
703 }
704 return filter;
705 }
706
707 /**
708 * @return the associated paths associated with this Destination
709 */
710 public String[] getDestinationPaths() {
711 if (paths == null) {
712 paths = DestinationPath.getDestinationPaths(physicalName);
713 }
714 return paths;
715 }
716
717 /**
718 * @return stats for this destination
719 */
720 public JMSDestinationStats getStats() {
721 if (stats == null) {
722 stats = createDestinationStats();
723 }
724 return stats;
725 }
726
727
728 /**
729 * @param stats
730 */
731 public void setStats(JMSDestinationStats stats) {
732 this.stats = stats;
733 }
734
735 /**
736 * increment counter for number of interested consumers
737 */
738 synchronized public void incrementConsumerCounter() {
739 consumerCounter++;
740 }
741
742 /**
743 * descrement counter for number interested consumers
744 */
745 synchronized public void decrementConsumerCounter() {
746 consumerCounter--;
747 }
748
749 /**
750 * @return true if this destination is deleted
751 */
752 synchronized public boolean isDeleted() {
753 return deleted;
754 }
755
756 /**
757 * det the deleted flag to the new value
758 * @param value
759 */
760 synchronized public void setDeleted(boolean value){
761 deleted = value;
762 }
763
764 /**
765 * Used to Deletes a temporary destination. If there are existing consumers
766 * still using it, a <CODE>JMSException</CODE> will be thrown.
767 *
768 * @throws JMSException if the JMS provider fails to delete the
769 * temporary queue due to some internal error.
770 */
771 synchronized public void delete() throws JMSException {
772 if (consumerCounter != 0) {
773 throw new JMSException("A consumer is still using this temporary queue.");
774 }
775 if (sessionCreatedBy != null) {
776 sessionCreatedBy.removeTemporaryDestination(this);
777 }
778 deleted = true;
779 }
780
781
782 // Implementation methods
783 //-------------------------------------------------------------------------
784
785
786 /**
787 * Factory method to create a child destination if this destination is a composite
788 * @param name
789 * @return the created Destination
790 */
791 protected abstract Destination createDestination(String name);
792
793 /**
794 * Factory method to create a statistics counter object
795 *
796 * @return
797 */
798 protected abstract JMSDestinationStats createDestinationStats();
799
800 /**
801 * Set the properties that will represent the instance in JNDI
802 *
803 * @param props
804 */
805 protected void buildFromProperties(Properties props) {
806 this.physicalName = props.getProperty("physicalName", this.physicalName);
807
808 }
809
810 /**
811 * Initialize the instance from properties stored in JNDI
812 *
813 * @param props
814 */
815
816 protected void populateProperties(Properties props) {
817 props.put("physicalName", this.physicalName);
818 }
819
820 /**
821 * @return the topic that is used for this destination's temporary destination advisories.
822 */
823 public ActiveMQTopic getTopicForTempAdvisory() {
824 String destName = ActiveMQDestination.TEMP_DESTINATION_ADVISORY_PREFIX+getAdvisoryDestinationTypePrefix()+getPhysicalName();
825 return new ActiveMQTopic(destName);
826 }
827
828 /**
829 * @return the topic that is used for this destination's consumer advisories.
830 */
831 public ActiveMQTopic getTopicForConsumerAdvisory() {
832 String destName = ActiveMQDestination.CONSUMER_ADVISORY_PREFIX+getAdvisoryDestinationTypePrefix()+getPhysicalName();
833 return new ActiveMQTopic(destName);
834 }
835
836 /**
837 * @return the topic that is used for this destination's producer advisories.
838 */
839 public ActiveMQTopic getTopicForProducerAdvisory() {
840 String destName = ActiveMQDestination.PRODUCER_ADVISORY_PREFIX+getAdvisoryDestinationTypePrefix()+getPhysicalName();
841 return new ActiveMQTopic(destName);
842 }
843
844 /**
845 * @return null if this destination is not an advisory topic, else it returns the
846 * destination that the advisories are related.
847 */
848 public ActiveMQDestination getDestinationBeingAdvised() {
849 if( isConnectionAdvisory() ) {
850 return null;
851 } else if( isConsumerAdvisory() ) {
852 String matchName = physicalName.substring(ActiveMQDestination.CONSUMER_ADVISORY_PREFIX.length());
853 return createDestinationFromAdvisoryName(matchName);
854 } else if ( isProducerAdvisory() ) {
855 String matchName = physicalName.substring(ActiveMQDestination.PRODUCER_ADVISORY_PREFIX.length());
856 return createDestinationFromAdvisoryName(matchName);
857 } else if ( isTempDestinationAdvisory() ) {
858 String matchName = physicalName.substring(ActiveMQDestination.TEMP_DESTINATION_ADVISORY_PREFIX.length());
859 return createDestinationFromAdvisoryName(matchName);
860 }
861 return null;
862 }
863
864 private String getAdvisoryDestinationTypePrefix() {
865 switch( getDestinationType() ) {
866 case ACTIVEMQ_TOPIC:
867 return "topic.";
868 case ACTIVEMQ_QUEUE:
869 return "queue.";
870 case ACTIVEMQ_TEMPORARY_TOPIC:
871 return "temp-topic.";
872 case ACTIVEMQ_TEMPORARY_QUEUE:
873 return "temp-queue.";
874 }
875 throw new RuntimeException("Invlaid destiantion type: "+getDestinationType());
876 }
877
878 /**
879 * @param advisoryName
880 * @return
881 */
882 private ActiveMQDestination createDestinationFromAdvisoryName(String advisoryName) {
883 if( advisoryName.startsWith("topic.") ) {
884 String name = advisoryName.substring("topic.".length());
885 return createDestination(ACTIVEMQ_TOPIC, name);
886 } else if( advisoryName.startsWith("queue.") ) {
887 String name = advisoryName.substring("queue.".length());
888 return createDestination(ACTIVEMQ_QUEUE, name);
889 } else if( advisoryName.startsWith("temp-topic.") ) {
890 String name = advisoryName.substring("temp-topic.".length());
891 return createDestination(ACTIVEMQ_TEMPORARY_TOPIC, name);
892 } else if( advisoryName.startsWith("temp-queue.") ) {
893 String name = advisoryName.substring("temp-queue.".length());
894 return createDestination(ACTIVEMQ_TEMPORARY_QUEUE, name);
895 }
896 return null;
897 }
898 }