QPID-2379: add Binding.msgMatched() support to the HeadersExchange

git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk@924879 13f79535-47bb-0310-9956-ffa450edef68

3 files changed, 128 insertions, 63 deletions

diff --git a/qpid/java/broker/src/test/java/org/apache/qpid/server/exchange/AbstractHeadersExchangeTestBase.java b/qpid/java/broker/src/test/java/org/apache/qpid/server/exchange/AbstractHeadersExchangeTestBase.java
index 06d5d80ac1..2d8b157297 100644
--- a/qpid/java/broker/src/test/java/org/apache/qpid/server/exchange/AbstractHeadersExchangeTestBase.java
+++ b/qpid/java/broker/src/test/java/org/apache/qpid/server/exchange/AbstractHeadersExchangeTestBase.java
@@ -31,6 +31,7 @@ import org.apache.qpid.framing.ContentHeaderBody;
 import org.apache.qpid.framing.FieldTable;
 import org.apache.qpid.framing.FieldTableFactory;
 import org.apache.qpid.framing.abstraction.MessagePublishInfo;
+import org.apache.qpid.server.binding.Binding;
 import org.apache.qpid.server.binding.BindingFactory;
 import org.apache.qpid.server.message.AMQMessage;
 import org.apache.qpid.server.message.AMQMessageHeader;
@@ -53,8 +54,10 @@ import org.apache.qpid.server.subscription.Subscription;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
+import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
+import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.atomic.AtomicLong;
 
@@ -92,31 +95,31 @@ public class AbstractHeadersExchangeTestBase extends TestCase
 
     protected TestQueue bindDefault(String... bindings) throws AMQException
     {
-        return bind("Queue" + (++count), bindings);
-    }
+        String queueName = "Queue" + (++count);
 
-    protected TestQueue bind(String queueName, String... bindings) throws AMQException
-    {
-        return bind(queueName, getHeaders(bindings));
+        return bind(queueName, queueName, getHeadersMap(bindings));
     }
-
-    protected TestQueue bind(String queue, FieldTable bindings) throws AMQException
+    
+    protected void unbind(TestQueue queue, String... bindings) throws AMQException
     {
-        return bind(new TestQueue(new AMQShortString(queue)), bindings);
+        String queueName = queue.getName();
+        //TODO - check this
+        exchange.onUnbind(new Binding(null,queueName, queue, exchange, getHeadersMap(bindings)));
     }
-
-    protected TestQueue bind(TestQueue queue, String... bindings) throws AMQException
+    
+    protected int getCount()
     {
-        return bind(queue, getHeaders(bindings));
+        return count;
     }
 
-    protected TestQueue bind(TestQueue queue, FieldTable bindings) throws AMQException
+    private TestQueue bind(String key, String queueName, Map<String,Object> args) throws AMQException
     {
+        TestQueue queue = new TestQueue(new AMQShortString(queueName));
         queues.add(queue);
-        exchange.registerQueue(null, queue, bindings);
+        exchange.onBind(new Binding(null,key, queue, exchange, args));
         return queue;
     }
-
+    
 
     protected int route(Message m) throws AMQException
     {
@@ -171,6 +174,23 @@ public class AbstractHeadersExchangeTestBase extends TestCase
             }
 
     }
+    
+    static Map<String,Object> getHeadersMap(String... entries)
+    {
+        if(entries == null)
+        {
+            return null;
+        }
+        
+        Map<String,Object> headers = new HashMap<String,Object>();
+
+        for (String s : entries)
+        {
+            String[] parts = s.split("=", 2);
+            headers.put(parts[0], parts.length > 1 ? parts[1] : "");
+        }
+        return headers;
+    }
 
     static FieldTable getHeaders(String... entries)
     {
diff --git a/qpid/java/broker/src/test/java/org/apache/qpid/server/exchange/HeadersBindingTest.java b/qpid/java/broker/src/test/java/org/apache/qpid/server/exchange/HeadersBindingTest.java
index 1e56a32383..a7c226cbd8 100644
--- a/qpid/java/broker/src/test/java/org/apache/qpid/server/exchange/HeadersBindingTest.java
+++ b/qpid/java/broker/src/test/java/org/apache/qpid/server/exchange/HeadersBindingTest.java
@@ -26,7 +26,9 @@ import java.util.Set;
 
 import junit.framework.TestCase;
 import org.apache.qpid.framing.FieldTable;
+import org.apache.qpid.server.binding.Binding;
 import org.apache.qpid.server.message.AMQMessageHeader;
+import org.apache.qpid.server.queue.MockAMQQueue;
 
 /**
  */
@@ -119,166 +121,193 @@ public class HeadersBindingTest extends TestCase
         }
     }
 
-    private FieldTable bindHeaders = new FieldTable();
+    private Map<String,Object> bindHeaders = new HashMap<String,Object>();
     private MockHeader matchHeaders = new MockHeader();
+    private int _count = 0;
+    private MockAMQQueue _queue;
+    
+    protected void setUp()
+    {
+        _count++;
+        _queue = new MockAMQQueue(getQueueName());
+    }
+    
+    protected String getQueueName()
+    {
+        return "Queue" + _count;
+    }
 
     public void testDefault_1()
     {
-        bindHeaders.setString("A", "Value of A");
+        bindHeaders.put("A", "Value of A");
 
         matchHeaders.setString("A", "Value of A");
 
-        assertTrue(new HeadersBinding(bindHeaders).matches(matchHeaders));
+        Binding b = new Binding(null, getQueueName(), _queue, null, bindHeaders);
+        assertTrue(new HeadersBinding(b).matches(matchHeaders));
     }
 
     public void testDefault_2()
     {
-        bindHeaders.setString("A", "Value of A");
+        bindHeaders.put("A", "Value of A");
 
         matchHeaders.setString("A", "Value of A");
         matchHeaders.setString("B", "Value of B");
 
-        assertTrue(new HeadersBinding(bindHeaders).matches(matchHeaders));
+        Binding b = new Binding(null, getQueueName(), _queue, null, bindHeaders);
+        assertTrue(new HeadersBinding(b).matches(matchHeaders));
     }
 
     public void testDefault_3()
     {
-        bindHeaders.setString("A", "Value of A");
+        bindHeaders.put("A", "Value of A");
 
         matchHeaders.setString("A", "Altered value of A");
 
-        assertFalse(new HeadersBinding(bindHeaders).matches(matchHeaders));
+        Binding b = new Binding(null, getQueueName(), _queue, null, bindHeaders);
+        assertFalse(new HeadersBinding(b).matches(matchHeaders));
     }
 
     public void testAll_1()
     {
-        bindHeaders.setString("X-match", "all");
-        bindHeaders.setString("A", "Value of A");
+        bindHeaders.put("X-match", "all");
+        bindHeaders.put("A", "Value of A");
 
         matchHeaders.setString("A", "Value of A");
 
-        assertTrue(new HeadersBinding(bindHeaders).matches(matchHeaders));
+        Binding b = new Binding(null, getQueueName(), _queue, null, bindHeaders);
+        assertTrue(new HeadersBinding(b).matches(matchHeaders));
     }
 
     public void testAll_2()
     {
-        bindHeaders.setString("X-match", "all");
-        bindHeaders.setString("A", "Value of A");
-        bindHeaders.setString("B", "Value of B");
+        bindHeaders.put("X-match", "all");
+        bindHeaders.put("A", "Value of A");
+        bindHeaders.put("B", "Value of B");
 
         matchHeaders.setString("A", "Value of A");
 
-        assertFalse(new HeadersBinding(bindHeaders).matches(matchHeaders));
+        Binding b = new Binding(null, getQueueName(), _queue, null, bindHeaders);
+        assertFalse(new HeadersBinding(b).matches(matchHeaders));
     }
 
     public void testAll_3()
     {
-        bindHeaders.setString("X-match", "all");
-        bindHeaders.setString("A", "Value of A");
-        bindHeaders.setString("B", "Value of B");
+        bindHeaders.put("X-match", "all");
+        bindHeaders.put("A", "Value of A");
+        bindHeaders.put("B", "Value of B");
 
         matchHeaders.setString("A", "Value of A");
         matchHeaders.setString("B", "Value of B");
 
-        assertTrue(new HeadersBinding(bindHeaders).matches(matchHeaders));
+        Binding b = new Binding(null, getQueueName(), _queue, null, bindHeaders);
+        assertTrue(new HeadersBinding(b).matches(matchHeaders));
     }
 
     public void testAll_4()
     {
-        bindHeaders.setString("X-match", "all");
-        bindHeaders.setString("A", "Value of A");
-        bindHeaders.setString("B", "Value of B");
+        bindHeaders.put("X-match", "all");
+        bindHeaders.put("A", "Value of A");
+        bindHeaders.put("B", "Value of B");
 
         matchHeaders.setString("A", "Value of A");
         matchHeaders.setString("B", "Value of B");
         matchHeaders.setString("C", "Value of C");
 
-        assertTrue(new HeadersBinding(bindHeaders).matches(matchHeaders));
+        Binding b = new Binding(null, getQueueName(), _queue, null, bindHeaders);
+        assertTrue(new HeadersBinding(b).matches(matchHeaders));
     }
 
     public void testAll_5()
     {
-        bindHeaders.setString("X-match", "all");
-        bindHeaders.setString("A", "Value of A");
-        bindHeaders.setString("B", "Value of B");
+        bindHeaders.put("X-match", "all");
+        bindHeaders.put("A", "Value of A");
+        bindHeaders.put("B", "Value of B");
 
         matchHeaders.setString("A", "Value of A");
         matchHeaders.setString("B", "Altered value of B");
         matchHeaders.setString("C", "Value of C");
 
-        assertFalse(new HeadersBinding(bindHeaders).matches(matchHeaders));
+        Binding b = new Binding(null, getQueueName(), _queue, null, bindHeaders);
+        assertFalse(new HeadersBinding(b).matches(matchHeaders));
     }
 
     public void testAny_1()
     {
-        bindHeaders.setString("X-match", "any");
-        bindHeaders.setString("A", "Value of A");
+        bindHeaders.put("X-match", "any");
+        bindHeaders.put("A", "Value of A");
 
         matchHeaders.setString("A", "Value of A");
 
-        assertTrue(new HeadersBinding(bindHeaders).matches(matchHeaders));
+        Binding b = new Binding(null, getQueueName(), _queue, null, bindHeaders);
+        assertTrue(new HeadersBinding(b).matches(matchHeaders));
     }
 
     public void testAny_2()
     {
-        bindHeaders.setString("X-match", "any");
-        bindHeaders.setString("A", "Value of A");
-        bindHeaders.setString("B", "Value of B");
+        bindHeaders.put("X-match", "any");
+        bindHeaders.put("A", "Value of A");
+        bindHeaders.put("B", "Value of B");
 
         matchHeaders.setString("A", "Value of A");
 
-        assertTrue(new HeadersBinding(bindHeaders).matches(matchHeaders));
+        Binding b = new Binding(null, getQueueName(), _queue, null, bindHeaders);
+        assertTrue(new HeadersBinding(b).matches(matchHeaders));
     }
 
     public void testAny_3()
     {
-        bindHeaders.setString("X-match", "any");
-        bindHeaders.setString("A", "Value of A");
-        bindHeaders.setString("B", "Value of B");
+        bindHeaders.put("X-match", "any");
+        bindHeaders.put("A", "Value of A");
+        bindHeaders.put("B", "Value of B");
 
         matchHeaders.setString("A", "Value of A");
         matchHeaders.setString("B", "Value of B");
 
-        assertTrue(new HeadersBinding(bindHeaders).matches(matchHeaders));
+        Binding b = new Binding(null, getQueueName(), _queue, null, bindHeaders);
+        assertTrue(new HeadersBinding(b).matches(matchHeaders));
     }
 
     public void testAny_4()
     {
-        bindHeaders.setString("X-match", "any");
-        bindHeaders.setString("A", "Value of A");
-        bindHeaders.setString("B", "Value of B");
+        bindHeaders.put("X-match", "any");
+        bindHeaders.put("A", "Value of A");
+        bindHeaders.put("B", "Value of B");
 
         matchHeaders.setString("A", "Value of A");
         matchHeaders.setString("B", "Value of B");
         matchHeaders.setString("C", "Value of C");
 
-        assertTrue(new HeadersBinding(bindHeaders).matches(matchHeaders));
+        Binding b = new Binding(null, getQueueName(), _queue, null, bindHeaders);
+        assertTrue(new HeadersBinding(b).matches(matchHeaders));
     }
 
     public void testAny_5()
     {
-        bindHeaders.setString("X-match", "any");
-        bindHeaders.setString("A", "Value of A");
-        bindHeaders.setString("B", "Value of B");
+        bindHeaders.put("X-match", "any");
+        bindHeaders.put("A", "Value of A");
+        bindHeaders.put("B", "Value of B");
 
         matchHeaders.setString("A", "Value of A");
         matchHeaders.setString("B", "Altered value of B");
         matchHeaders.setString("C", "Value of C");
 
-        assertTrue(new HeadersBinding(bindHeaders).matches(matchHeaders));
+        Binding b = new Binding(null, getQueueName(), _queue, null, bindHeaders);
+        assertTrue(new HeadersBinding(b).matches(matchHeaders));
     }
 
     public void testAny_6()
     {
-        bindHeaders.setString("X-match", "any");
-        bindHeaders.setString("A", "Value of A");
-        bindHeaders.setString("B", "Value of B");
+        bindHeaders.put("X-match", "any");
+        bindHeaders.put("A", "Value of A");
+        bindHeaders.put("B", "Value of B");
 
         matchHeaders.setString("A", "Altered value of A");
         matchHeaders.setString("B", "Altered value of B");
         matchHeaders.setString("C", "Value of C");
 
-        assertFalse(new HeadersBinding(bindHeaders).matches(matchHeaders));
+        Binding b = new Binding(null, getQueueName(), _queue, null, bindHeaders);
+        assertFalse(new HeadersBinding(b).matches(matchHeaders));
     }
 
     public static junit.framework.Test suite()
diff --git a/qpid/java/broker/src/test/java/org/apache/qpid/server/exchange/HeadersExchangeTest.java b/qpid/java/broker/src/test/java/org/apache/qpid/server/exchange/HeadersExchangeTest.java
index 580bc78b8d..f982c3976f 100644
--- a/qpid/java/broker/src/test/java/org/apache/qpid/server/exchange/HeadersExchangeTest.java
+++ b/qpid/java/broker/src/test/java/org/apache/qpid/server/exchange/HeadersExchangeTest.java
@@ -106,6 +106,22 @@ public class HeadersExchangeTest extends AbstractHeadersExchangeTestBase
         pb2.setMandatory(true);
         routeAndTest(m1,true);
     }
+    
+    public void testOnUnbind() throws AMQException
+    {
+        TestQueue q1 = bindDefault("F0000");
+        TestQueue q2 = bindDefault("F0000=Aardvark");
+        TestQueue q3 = bindDefault("F0001");
+        
+        routeAndTest(new Message(_protocolSession, "Message1", "F0000"), q1);
+        routeAndTest(new Message(_protocolSession, "Message2", "F0000=Aardvark"), q1, q2);
+        routeAndTest(new Message(_protocolSession, "Message3", "F0001"), q3);
+        
+        unbind(q1,"F0000");
+        routeAndTest(new Message(_protocolSession, "Message4", "F0000"));
+        routeAndTest(new Message(_protocolSession, "Message5", "F0000=Aardvark"), q2);
+    }
+    
 
     public static junit.framework.Test suite()
     {