svn commit: r753070 - in /activemq/sandbox/activemq-flow/src/main/java/org/apache/activemq/broker: Domain.java QueueDomain.java Router.java TopicDomain.java

[ch...@apache.org]

Author: chirino
Date: Fri Mar 13 00:08:47 2009
New Revision: 753070

URL: http://svn.apache.org/viewvc?rev=753070&view=rev
Log:
Add routing support for composite/multi destinations

Modified:
    activemq/sandbox/activemq-flow/src/main/java/org/apache/activemq/broker/Domain.java
    activemq/sandbox/activemq-flow/src/main/java/org/apache/activemq/broker/QueueDomain.java
    activemq/sandbox/activemq-flow/src/main/java/org/apache/activemq/broker/Router.java
    activemq/sandbox/activemq-flow/src/main/java/org/apache/activemq/broker/TopicDomain.java

Modified: activemq/sandbox/activemq-flow/src/main/java/org/apache/activemq/broker/Domain.java
URL: http://svn.apache.org/viewvc/activemq/sandbox/activemq-flow/src/main/java/org/apache/activemq/broker/Domain.java?rev=753070&r1=753069&r2=753070&view=diff
==============================================================================
--- activemq/sandbox/activemq-flow/src/main/java/org/apache/activemq/broker/Domain.java (original)
+++ activemq/sandbox/activemq-flow/src/main/java/org/apache/activemq/broker/Domain.java Fri Mar 13 00:08:47 2009
@@ -12,12 +12,12 @@
  */
 public interface Domain {
 
-    public void add(AsciiBuffer name, Object value);
+    public void add(AsciiBuffer destinationName, Object destination);
     
-    public Object remove(AsciiBuffer name);
+    public Object remove(AsciiBuffer destinationName);
 
-    public void bind(AsciiBuffer name, DeliveryTarget dt);
+    public void bind(AsciiBuffer destinationName, DeliveryTarget deliveryTarget);
 
-    public Collection<DeliveryTarget> route(MessageDelivery msg);
+    public Collection<DeliveryTarget> route(AsciiBuffer destinationName, MessageDelivery message);
     
 }

Modified: activemq/sandbox/activemq-flow/src/main/java/org/apache/activemq/broker/QueueDomain.java
URL: http://svn.apache.org/viewvc/activemq/sandbox/activemq-flow/src/main/java/org/apache/activemq/broker/QueueDomain.java?rev=753070&r1=753069&r2=753070&view=diff
==============================================================================
--- activemq/sandbox/activemq-flow/src/main/java/org/apache/activemq/broker/QueueDomain.java (original)
+++ activemq/sandbox/activemq-flow/src/main/java/org/apache/activemq/broker/QueueDomain.java Fri Mar 13 00:08:47 2009
@@ -21,8 +21,8 @@
         queues.get(name).addConsumer(deliveryTarget);
     }
 
-    public Collection<DeliveryTarget> route(MessageDelivery delivery) {
-        Queue queue = queues.get(delivery.getDestination().getName());
+    public Collection<DeliveryTarget> route(AsciiBuffer name, MessageDelivery delivery) {
+        Queue queue = queues.get(name);
         if( queue!=null ) {
             ArrayList<DeliveryTarget> rc = new ArrayList<DeliveryTarget>(1);
             rc.add(queue);

Modified: activemq/sandbox/activemq-flow/src/main/java/org/apache/activemq/broker/Router.java
URL: http://svn.apache.org/viewvc/activemq/sandbox/activemq-flow/src/main/java/org/apache/activemq/broker/Router.java?rev=753070&r1=753069&r2=753070&view=diff
==============================================================================
--- activemq/sandbox/activemq-flow/src/main/java/org/apache/activemq/broker/Router.java (original)
+++ activemq/sandbox/activemq-flow/src/main/java/org/apache/activemq/broker/Router.java Fri Mar 13 00:08:47 2009
@@ -5,6 +5,7 @@
 
 import java.util.Collection;
 import java.util.HashMap;
+import java.util.HashSet;
 
 import org.apache.activemq.protobuf.AsciiBuffer;
 
@@ -39,8 +40,25 @@
     }
 
     public Collection<DeliveryTarget> route(MessageDelivery msg) {
-        Domain domain = domains.get(msg.getDestination().getDomain());
-        return domain.route(msg);
+        return route(msg.getDestination(), msg);
+    }
+
+    private Collection<DeliveryTarget> route(Destination destination, MessageDelivery msg) {
+        // Handles routing to composite/multi destinations.
+        Collection<Destination> destinationList = destination.getDestinations();
+        if( destinationList == null ) {
+            Domain domain = domains.get(destination.getDomain());
+            return domain.route(destination.getName(), msg);
+        } else {
+            HashSet<DeliveryTarget> rc = new HashSet<DeliveryTarget>();
+            for (Destination d : destinationList) {
+                Collection<DeliveryTarget> t = route(d, msg);
+                if( t!=null ) {
+                    rc.addAll(t);
+                }
+            }            
+            return rc;
+        }
     }
 
 }
\ No newline at end of file

Modified: activemq/sandbox/activemq-flow/src/main/java/org/apache/activemq/broker/TopicDomain.java
URL: http://svn.apache.org/viewvc/activemq/sandbox/activemq-flow/src/main/java/org/apache/activemq/broker/TopicDomain.java?rev=753070&r1=753069&r2=753070&view=diff
==============================================================================
--- activemq/sandbox/activemq-flow/src/main/java/org/apache/activemq/broker/TopicDomain.java (original)
+++ activemq/sandbox/activemq-flow/src/main/java/org/apache/activemq/broker/TopicDomain.java Fri Mar 13 00:08:47 2009
@@ -25,8 +25,8 @@
         targets.add(target);
     }
 
-    public Collection<DeliveryTarget> route(MessageDelivery delivery) {
-        return topicsTargets.get(delivery.getDestination().getName());
+    public Collection<DeliveryTarget> route(AsciiBuffer name, MessageDelivery delivery) {
+        return topicsTargets.get(name);
     }
 
 }