QPID-2418: updates to fix test failures when using the 0-10 client test profiles.

Use a transacted session when querying for queue counts following consumption, as the 0-10 client batches auto-acks asynchronously.
Always send the selector filter argument even if empty, to allow querying the brokers via 0-10 to detect whether the selector is being added/removed/modified at subscribe time.
Enable the Java broker to perform argument matching during the 0-10 isBound check.


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

3 files changed, 25 insertions, 18 deletions

diff --git a/qpid/java/systests/src/main/java/org/apache/qpid/server/logging/BindingLoggingTest.java b/qpid/java/systests/src/main/java/org/apache/qpid/server/logging/BindingLoggingTest.java
index afda7d4ba9..51815e2adc 100644
--- a/qpid/java/systests/src/main/java/org/apache/qpid/server/logging/BindingLoggingTest.java
+++ b/qpid/java/systests/src/main/java/org/apache/qpid/server/logging/BindingLoggingTest.java
@@ -94,7 +94,7 @@ public class BindingLoggingTest extends AbstractTestLogging
      * 2. New Client requests that a Queue is bound to a new exchange.
      * Output:
      *
-     * <date> BND-1001 : Create
+     * <date> BND-1001 : Create :  Arguments : {x-filter-jms-selector=}
      *
      * Validation Steps:
      * 3. The BND ID is correct
@@ -117,6 +117,7 @@ public class BindingLoggingTest extends AbstractTestLogging
         validateLogMessage(getLogMessage(results, 0), messageID, message, exchange, queueName, queueName);
 
         exchange = "direct/amq.direct";
+        message = "Create : Arguments : {x-filter-jms-selector=}";
         validateLogMessage(getLogMessage(results, 1), messageID, message, exchange, queueName, queueName);
     }
 
@@ -129,7 +130,7 @@ public class BindingLoggingTest extends AbstractTestLogging
      * 2. Java Client consumes from a topic with a JMS selector.
      * Output:
      *
-     * <date> BND-1001 : Create : Arguments : <key=value>
+     * <date> BND-1001 : Create : Arguments : {x-filter-jms-selector=<value>}
      *
      * Validation Steps:
      * 3. The BND ID is correct
diff --git a/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/message/StreamMessageTest.java b/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/message/StreamMessageTest.java
index 98d59982e5..0f799073b4 100644
--- a/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/message/StreamMessageTest.java
+++ b/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/message/StreamMessageTest.java
@@ -75,6 +75,7 @@ public class StreamMessageTest extends QpidBrokerTestCase
                     ExchangeDefaults.HEADERS_EXCHANGE_CLASS + "://" + ExchangeDefaults.HEADERS_EXCHANGE_NAME
                     + "/test/queue1?" + BindingURL.OPTION_ROUTING_KEY + "='F0000=1'"));
         FieldTable ft = new FieldTable();
+        ft.setString("x-match", "any");
         ft.setString("F1000", "1");
         MessageConsumer consumer =
             consumerSession.createConsumer(queue, Integer.parseInt(ClientProperties.MAX_PREFETCH_DEFAULT), Integer.parseInt(ClientProperties.MAX_PREFETCH_DEFAULT), false, false, (String) null, ft);
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 3dd3c72024..d73761d12a 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
@@ -544,23 +544,22 @@ public class DurableSubscriptionTest extends QpidBrokerTestCase
         TopicSubscriber subB = session.createDurableSubscriber(topic, 
                 "testResubscribeWithChangedSelector","Match = False", false);
 
-        //verify no messages are now present as changing selector should have issued
-        //an unsubscribe and thus deleted the previous backing queue for the subscription.
+        //verify no messages are now recieved.
         rMsg = subB.receive(NEGATIVE_RECEIVE_TIMEOUT);
-        assertNull("Should not have received message as the queue underlying the " +
-                 "subscription should have been cleared/deleted when the selector was changed", rMsg);
+        assertNull("Should not have received message as the selector was changed", rMsg);
 
         // Check that new messages are received properly
         sendMatchingAndNonMatchingMessage(session, producer);
         rMsg = subB.receive(POSITIVE_RECEIVE_TIMEOUT);
 
-        assertNotNull("Message should not be received", rMsg);
+        assertNotNull("Message should have been received", rMsg);
         assertEquals("Content was wrong", 
                      "testResubscribeWithChangedSelector2",
                      ((TextMessage) rMsg).getText());
         
+        
         rMsg = subB.receive(NEGATIVE_RECEIVE_TIMEOUT);
-        assertNull("Message should be received",rMsg);
+        assertNull("Message should not have been received",rMsg);
         session.unsubscribe("testResubscribeWithChangedSelector");
     }
 
@@ -613,7 +612,7 @@ public class DurableSubscriptionTest extends QpidBrokerTestCase
     {        
         Connection conn = getConnection();
         conn.start();
-        Session session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
+        Session session = conn.createSession(true, Session.SESSION_TRANSACTED);
         AMQTopic topic = new AMQTopic((AMQConnection) conn, "sameMessageSelector");
                 
         //create and register a durable subscriber with a message selector and then close it
@@ -630,15 +629,17 @@ public class DurableSubscriptionTest extends QpidBrokerTestCase
             message.setBooleanProperty("testprop", false);
             producer.send(message);
         }
+        session.commit();
         producer.close();
 
+        // should be 5 or 10 messages on queue now
+        // (5 for the java broker due to use of server side selectors, and 10 for the cpp broker due to client side selectors only)
+        AMQQueue queue = new AMQQueue("amq.topic", "clientid" + ":" + "sameMessageSelector");
+        assertEquals("Queue depth is wrong", isJavaBroker() ? 5 : 10, ((AMQSession<?, ?>) session).getQueueDepth(queue));
+
         // now recreate the durable subscriber and check the received messages
         TopicSubscriber subTwo = session.createDurableSubscriber(topic, "sameMessageSelector", "testprop = TRUE", false);
 
-        // should be 5 messages on queue now
-        AMQQueue queue = new AMQQueue("amq.topic", "clientid" + ":" + "sameMessageSelector");
-        assertEquals("Queue depth is wrong", 5, ((AMQSession<?, ?>) session).getQueueDepth(queue));
-        
         for (int i = 0; i < 5; i++)
         {
             Message message = subTwo.receive(1000);
@@ -653,6 +654,8 @@ public class DurableSubscriptionTest extends QpidBrokerTestCase
             }
         }
         
+        session.commit();
+        
         // Check queue has no messages
         assertEquals("Queue should be empty", 0, ((AMQSession<?, ?>) session).getQueueDepth(queue));
         
@@ -710,10 +713,11 @@ public class DurableSubscriptionTest extends QpidBrokerTestCase
         msg.setBooleanProperty("Match", false);
         producer.send(msg);
 
-        // should be 1 message on queue now
+        // should be 1 or 2 messages on queue now
+        // (1 for the java broker due to use of server side selectors, and 2 for the cpp broker due to client side selectors only)
         AMQQueue queue = new AMQQueue("amq.topic", "clientid" + ":" + "testResubscribeWithChangedSelectorNoClose");
-        assertEquals("Queue depth is wrong", 1, ((AMQSession<?, ?>) session).getQueueDepth(queue));
-        
+        assertEquals("Queue depth is wrong", isJavaBroker() ? 1 : 2, ((AMQSession<?, ?>) session).getQueueDepth(queue));
+
         Message rMsg = subB.receive(1000);
         assertNotNull(rMsg);
         assertEquals("Content was wrong", 
@@ -773,9 +777,10 @@ public class DurableSubscriptionTest extends QpidBrokerTestCase
         msg.setBooleanProperty("testprop", false);
         producer.send(msg);
 
-        // should be 1 message on queue now
+        // should be 1 or 2 messages on queue now
+        // (1 for the java broker due to use of server side selectors, and 2 for the cpp broker due to client side selectors only)
         AMQQueue queue = new AMQQueue("amq.topic", "clientid" + ":" + "subscriptionName");
-        assertEquals("Queue depth is wrong", 1, ((AMQSession<?, ?>) session).getQueueDepth(queue));
+        assertEquals("Queue depth is wrong", isJavaBroker() ? 1 : 2, ((AMQSession<?, ?>) session).getQueueDepth(queue));
         
         Message rMsg = subTwo.receive(1000);
         assertNotNull(rMsg);