QPID-1192: Make consumer send Selector as part of binding.

QPID-1191: Add test to exhibit leak

Change DurableSubscriptionTest to validate exception type recieved
Make BasicMessageConsumer validate the Selector before attempting creation

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

2 files changed, 51 insertions, 3 deletions

diff --git a/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/topic/DurableSubscriptionTest.java b/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/topic/DurableSubscriptionTest.java
index 4897f5fa15..2d3a98977f 100644
--- a/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/topic/DurableSubscriptionTest.java
+++ b/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/topic/DurableSubscriptionTest.java
@@ -31,6 +31,7 @@ import org.slf4j.LoggerFactory;
 import javax.jms.Connection;
 import javax.jms.InvalidDestinationException;
 import javax.jms.InvalidSelectorException;
+import javax.jms.JMSException;
 import javax.jms.Message;
 import javax.jms.MessageConsumer;
 import javax.jms.MessageProducer;
@@ -290,9 +291,9 @@ public class DurableSubscriptionTest extends QpidTestCase
 																	 		 "=TEST 'test", true);
     		assertNull("Subscriber should not have been created", deadSubscriber);
     	} 
-    	catch (InvalidSelectorException e)
+    	catch (JMSException e)
     	{
-    		// This was expected
+    		assertTrue("Wrong type of exception thrown", e instanceof InvalidSelectorException);
     	}
     	
     	TopicSubscriber liveSubscriber = session.createDurableSubscriber(topic, "testDurableWithInvalidSelectorSub");
diff --git a/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/topic/TopicSessionTest.java b/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/topic/TopicSessionTest.java
index 65e0563141..7a20b1058b 100644
--- a/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/topic/TopicSessionTest.java
+++ b/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/topic/TopicSessionTest.java
@@ -33,6 +33,7 @@ import javax.jms.TopicSubscriber;
 import org.apache.qpid.client.AMQConnection;
 import org.apache.qpid.client.AMQSession;
 import org.apache.qpid.client.AMQTopic;
+import org.apache.qpid.client.AMQTopicSessionAdaptor;
 import org.apache.qpid.test.utils.QpidTestCase;
 
 
@@ -319,7 +320,7 @@ public class TopicSessionTest extends QpidTestCase
         assertNull(m);
 
         //send message to all consumers
-         message = session1.createTextMessage("hello2");
+        message = session1.createTextMessage("hello2");
         message.setStringProperty("Selector", "select");
 
         publisher.publish(message);
@@ -362,6 +363,52 @@ public class TopicSessionTest extends QpidTestCase
         con.close();
         con2.close();
     }
+    
+    /**
+     * This tests QPID-1191, where messages which are sent to a topic but are not consumed by a subscriber
+     * due to a selector can be leaked.
+     * @throws Exception 
+     */
+    public void testNonMatchingMessagesDoNotFillQueue() throws Exception
+    {
+        AMQConnection con = (AMQConnection) getConnection("guest", "guest");
+
+        // Setup Topic
+        AMQTopic topic = new AMQTopic(con, "testNoLocal");
+
+        TopicSession session = con.createTopicSession(false, AMQSession.NO_ACKNOWLEDGE);
+
+        // Setup subscriber with selector
+        TopicSubscriber selector = session.createSubscriber(topic,  "Selector = 'select'", false);
+        TopicPublisher publisher = session.createPublisher(topic);
+
+        con.start();
+        TextMessage m;
+        TextMessage message;
+
+        // Send non-matching message
+        message = session.createTextMessage("non-matching 1");
+        publisher.publish(message);
+        
+        // Send and consume matching message
+        message = session.createTextMessage("hello");
+        message.setStringProperty("Selector", "select");
+
+        publisher.publish(message);
+
+        m = (TextMessage) selector.receive(1000);
+        assertNotNull("should have received message", m);
+        assertEquals("Message contents were wrong", "hello", m.getText());
+        
+        // Send non-matching message
+        message = session.createTextMessage("non-matching 2");
+        publisher.publish(message);
+
+        // Assert queue count is 0
+        long depth = ((AMQTopicSessionAdaptor) session).getSession().getQueueDepth(topic);
+        assertEquals("Queue depth was wrong", 0, depth);
+        
+    }
 
     public static junit.framework.Test suite()
     {