QPID-3642,QPID-3643: Add Dead Letter Queue functionality for 0-8/0-9/0-9-1 paths, fixes isBound methods on FanoutExchange

Applied patch from  Keith Wall <keith.wall@gmail.com>, Andrew MacBean <andymacbean@gmail.com> and Oleksandr Rudyy<orudyy@gmail.com>

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

5 files changed, 282 insertions, 11 deletions

diff --git a/qpid/java/client/src/test/java/org/apache/qpid/client/AMQConnectionUnitTest.java b/qpid/java/client/src/test/java/org/apache/qpid/client/AMQConnectionUnitTest.java
new file mode 100644
index 0000000000..3a565f0f0d
--- /dev/null
+++ b/qpid/java/client/src/test/java/org/apache/qpid/client/AMQConnectionUnitTest.java
@@ -0,0 +1,66 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+package org.apache.qpid.client;
+
+import java.util.concurrent.atomic.AtomicReference;
+
+import javax.jms.ExceptionListener;
+import javax.jms.JMSException;
+
+import junit.framework.TestCase;
+
+import org.apache.qpid.AMQInvalidArgumentException;
+
+public class AMQConnectionUnitTest extends TestCase
+{
+
+    public void testExceptionReceived()
+    {
+        String url = "amqp://guest:guest@/test?brokerlist='tcp://localhost:5672'";
+        AMQInvalidArgumentException expectedException = new AMQInvalidArgumentException("Test", null);
+        final AtomicReference<JMSException> receivedException = new AtomicReference<JMSException>();
+        try
+        {
+            MockAMQConnection connection = new MockAMQConnection(url);
+            connection.setExceptionListener(new ExceptionListener()
+            {
+
+                @Override
+                public void onException(JMSException jmsException)
+                {
+                    receivedException.set(jmsException);
+                }
+            });
+            connection.exceptionReceived(expectedException);
+        }
+        catch (Exception e)
+        {
+            fail("Failure to test exceptionRecived:" + e.getMessage());
+        }
+        JMSException exception = receivedException.get();
+        assertNotNull("Expected JMSException but got null", exception);
+        assertEquals("JMSException error code is incorrect", Integer.toString(expectedException.getErrorCode().getCode()), exception.getErrorCode());
+        assertNotNull("Expected not null message for JMSException", exception.getMessage());
+        assertTrue("JMSException error message is incorrect",  exception.getMessage().contains(expectedException.getMessage()));
+        assertEquals("JMSException linked exception is incorrect", expectedException, exception.getLinkedException());
+    }
+
+}
diff --git a/qpid/java/client/src/test/java/org/apache/qpid/client/BasicMessageConsumer_0_8_Test.java b/qpid/java/client/src/test/java/org/apache/qpid/client/BasicMessageConsumer_0_8_Test.java
new file mode 100644
index 0000000000..d8d94ba40e
--- /dev/null
+++ b/qpid/java/client/src/test/java/org/apache/qpid/client/BasicMessageConsumer_0_8_Test.java
@@ -0,0 +1,104 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+package org.apache.qpid.client;
+
+import javax.jms.Session;
+
+import org.apache.qpid.test.unit.message.TestAMQSession;
+import org.apache.qpid.url.AMQBindingURL;
+
+import junit.framework.TestCase;
+
+public class BasicMessageConsumer_0_8_Test extends TestCase
+{
+    /**
+     * Test that if there is a value for Reject Behaviour specified for the Destination
+     * used to create the Consumer, it overrides the value for the Connection.
+     */
+    public void testDestinationRejectBehaviourOverridesDefaultConnection() throws Exception
+    {
+        /*
+         * Check that when the connection does not have a value applied that this
+         * is successfully overridden with a specific value by the consumer.
+         */
+        String connUrlString = "amqp://guest:guest@/test?brokerlist='tcp://localhost:5672'";
+        AMQConnection conn = new MockAMQConnection(connUrlString);
+
+        String url = "exchangeClass://exchangeName/Destination/Queue?rejectbehaviour='server'";
+        AMQBindingURL burl = new AMQBindingURL(url);
+        AMQDestination queue = new AMQQueue(burl);
+
+        AMQSession<BasicMessageConsumer_0_8, BasicMessageProducer_0_8> testSession = new TestAMQSession(conn);
+        BasicMessageConsumer_0_8 consumer = new BasicMessageConsumer_0_8(0, conn, queue, "", false, null, testSession, null, null, 10, 5, false, Session.SESSION_TRANSACTED, false, false);
+
+        assertEquals("Reject behaviour was was not as expected", RejectBehaviour.SERVER, consumer.getRejectBehaviour());
+    }
+
+    /**
+     * Check that when the connection does have a specific value applied that this
+     * is successfully overridden with another specific value by the consumer.
+     */
+    public void testDestinationRejectBehaviourSpecified() throws Exception
+    {
+        final String connUrlString = "amqp://guest:guest@/test?brokerlist='tcp://localhost:5672'&rejectbehaviour='server'";
+        final AMQConnection conn = new MockAMQConnection(connUrlString);
+
+        final String url = "exchangeClass://exchangeName/Destination/Queue?rejectbehaviour='normal'";
+        final AMQBindingURL burl = new AMQBindingURL(url);
+        final AMQDestination queue = new AMQQueue(burl);
+
+        final AMQSession<BasicMessageConsumer_0_8, BasicMessageProducer_0_8> testSession = new TestAMQSession(conn);
+        final BasicMessageConsumer_0_8 consumer = new BasicMessageConsumer_0_8(0, conn, queue, "", false, null, testSession, null, null, 10, 5, false, Session.SESSION_TRANSACTED, false, false);
+
+        assertEquals("Reject behaviour was was not as expected", RejectBehaviour.NORMAL, consumer.getRejectBehaviour());
+    }
+
+    /**
+     * Test that if no value for Reject Behaviour is applied to the Destination, then the value
+     * from the connection is used and acts as expected.
+     */
+    public void testRejectBehaviourDetectedFromConnection() throws Exception
+    {
+        /*
+         * Check that when the connection does have a specific value applied that this
+         * is successfully detected by the consumer.
+         */
+        String connUrlString = "amqp://guest:guest@/test?brokerlist='tcp://localhost:5672'&rejectbehaviour='normal'";
+        AMQConnection conn = new MockAMQConnection(connUrlString);
+
+        String url = "exchangeClass://exchangeName/Destination/Queue";
+        AMQBindingURL burl = new AMQBindingURL(url);
+        AMQDestination queue = new AMQQueue(burl);
+
+        assertNull("Reject behaviour should have been null", queue.getRejectBehaviour());
+
+        AMQSession<BasicMessageConsumer_0_8, BasicMessageProducer_0_8> testSession = new TestAMQSession(conn);
+        BasicMessageConsumer_0_8 consumer = new BasicMessageConsumer_0_8(0, conn, queue, "", false, null, testSession, null, null, 10, 5, false, Session.SESSION_TRANSACTED, false, false);
+
+        assertEquals("Reject behaviour was was not as expected", RejectBehaviour.NORMAL, consumer.getRejectBehaviour());
+    }
+
+
+    protected RejectBehaviour getRejectBehaviour(AMQDestination destination)
+    {
+        return destination.getRejectBehaviour();
+    }
+}
diff --git a/qpid/java/client/src/test/java/org/apache/qpid/client/MockAMQConnection.java b/qpid/java/client/src/test/java/org/apache/qpid/client/MockAMQConnection.java
index 73e67469ae..919809edc3 100644
--- a/qpid/java/client/src/test/java/org/apache/qpid/client/MockAMQConnection.java
+++ b/qpid/java/client/src/test/java/org/apache/qpid/client/MockAMQConnection.java
@@ -55,4 +55,9 @@ public class MockAMQConnection extends AMQConnection
         _protocolHandler.getStateManager().changeState(AMQState.CONNECTION_OPEN);
         return null;
     }
+
+    public AMQConnectionDelegate getDelegate()
+    {
+        return _delegate;
+    }
 }
diff --git a/qpid/java/client/src/test/java/org/apache/qpid/test/unit/client/connectionurl/ConnectionURLTest.java b/qpid/java/client/src/test/java/org/apache/qpid/test/unit/client/connectionurl/ConnectionURLTest.java
index 4624b36fea..5a5a3a0bd9 100644
--- a/qpid/java/client/src/test/java/org/apache/qpid/test/unit/client/connectionurl/ConnectionURLTest.java
+++ b/qpid/java/client/src/test/java/org/apache/qpid/test/unit/client/connectionurl/ConnectionURLTest.java
@@ -7,9 +7,9 @@
  * to you under the Apache License, Version 2.0 (the
  * "License"); you may not use this file except in compliance
  * with the License.  You may obtain a copy of the License at
- * 
+ *
  *   http://www.apache.org/licenses/LICENSE-2.0
- * 
+ *
  * Unless required by applicable law or agreed to in writing,
  * software distributed under the License is distributed on an
  * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
@@ -38,7 +38,7 @@ public class ConnectionURLTest extends TestCase
         ConnectionURL connectionurl = new AMQConnectionURL(url);
 
         assertTrue(connectionurl.getFailoverMethod().equals("roundrobin"));
-        assertEquals("100", connectionurl.getFailoverOption(ConnectionURL.OPTIONS_FAILOVER_CYCLE));        
+        assertEquals("100", connectionurl.getFailoverOption(ConnectionURL.OPTIONS_FAILOVER_CYCLE));
         assertTrue(connectionurl.getUsername().equals("ritchiem"));
         assertTrue(connectionurl.getPassword().equals("bob"));
         assertTrue(connectionurl.getVirtualHost().equals("/test"));
@@ -338,7 +338,7 @@ public class ConnectionURLTest extends TestCase
         assertTrue(connectionurl.getPassword().equals("pass"));
         assertTrue(connectionurl.getVirtualHost().equals("/test"));
         assertTrue(connectionurl.getClientName().equals("client_id"));
-        
+
         assertTrue(connectionurl.getBrokerCount() == 1);
     }
 
@@ -457,7 +457,6 @@ public class ConnectionURLTest extends TestCase
 
         assertTrue(service.getTransport().equals("tcp"));
 
-        
         assertTrue(service.getHost().equals("localhost"));
         assertTrue(service.getPort() == 5672);
         assertEquals("jim",service.getProperty("foo"));
@@ -468,7 +467,7 @@ public class ConnectionURLTest extends TestCase
         assertTrue(connectionurl.getOption("timeout").equals("200"));
         assertTrue(connectionurl.getOption("immediatedelivery").equals("true"));
     }
-    
+
     /**
      * Test that options other than failover and brokerlist are returned in the string representation.
      * <p>
@@ -477,7 +476,7 @@ public class ConnectionURLTest extends TestCase
     public void testOptionToString() throws Exception
     {
         ConnectionURL url = new AMQConnectionURL("amqp://user:pass@temp/test?maxprefetch='12345'&brokerlist='tcp://localhost:5672'");
-        
+
         assertTrue("String representation should contain options and values", url.toString().contains("maxprefetch='12345'"));
     }
 
@@ -493,10 +492,10 @@ public class ConnectionURLTest extends TestCase
 
         assertTrue(connectionurl.getBrokerCount() == 1);
         BrokerDetails service = connectionurl.getBrokerDetails(0);
-        assertTrue(service.getTransport().equals("tcp"));        
+        assertTrue(service.getTransport().equals("tcp"));
         assertTrue(service.getHost().equals("under_score"));
         assertTrue(service.getPort() == 6672);
-        
+
         url = "amqp://guest:guest@clientid/test?brokerlist='tcp://under_score'";
 
         connectionurl = new AMQConnectionURL(url);
@@ -507,11 +506,44 @@ public class ConnectionURLTest extends TestCase
 
         assertTrue(connectionurl.getBrokerCount() == 1);
         service = connectionurl.getBrokerDetails(0);
-        assertTrue(service.getTransport().equals("tcp"));        
+        assertTrue(service.getTransport().equals("tcp"));
         assertTrue(service.getHost().equals("under_score"));
         assertTrue(service.getPort() == 5672);
     }
-    
+
+
+    public void testRejectBehaviourPresent() throws Exception
+    {
+        String url = "amqp://guest:guest@/test?brokerlist='tcp://localhost:5672'&rejectbehaviour='server'";
+
+        ConnectionURL connectionURL = new AMQConnectionURL(url);
+
+        assertTrue(connectionURL.getFailoverMethod() == null);
+        assertTrue(connectionURL.getUsername().equals("guest"));
+        assertTrue(connectionURL.getPassword().equals("guest"));
+        assertTrue(connectionURL.getVirtualHost().equals("/test"));
+
+        //check that the reject behaviour option is returned as expected
+        assertEquals("Reject behaviour option was not as expected", "server",
+                connectionURL.getOption(ConnectionURL.OPTIONS_REJECT_BEHAVIOUR));
+    }
+
+    public void testRejectBehaviourNotPresent() throws URLSyntaxException
+    {
+        String url = "amqp://guest:guest@/test?brokerlist='tcp://localhost:5672'&foo='bar'";
+
+        ConnectionURL connectionurl = new AMQConnectionURL(url);
+
+        assertTrue(connectionurl.getFailoverMethod() == null);
+        assertTrue(connectionurl.getUsername().equals("guest"));
+        assertTrue(connectionurl.getPassword().equals("guest"));
+        assertTrue(connectionurl.getVirtualHost().equals("/test"));
+
+        //check that the reject behaviour option is null as expected
+        assertNull("Reject behaviour option was not as expected",
+                connectionurl.getOption(ConnectionURL.OPTIONS_REJECT_BEHAVIOUR));
+    }
+
     public static junit.framework.Test suite()
     {
         return new junit.framework.TestSuite(ConnectionURLTest.class);
diff --git a/qpid/java/client/src/test/java/org/apache/qpid/test/unit/client/destinationurl/DestinationURLTest.java b/qpid/java/client/src/test/java/org/apache/qpid/test/unit/client/destinationurl/DestinationURLTest.java
index 7de09cff45..2c32e4c559 100644
--- a/qpid/java/client/src/test/java/org/apache/qpid/test/unit/client/destinationurl/DestinationURLTest.java
+++ b/qpid/java/client/src/test/java/org/apache/qpid/test/unit/client/destinationurl/DestinationURLTest.java
@@ -22,8 +22,11 @@ package org.apache.qpid.test.unit.client.destinationurl;
 
 import junit.framework.TestCase;
 
+import org.apache.qpid.client.AMQDestination;
+import org.apache.qpid.client.RejectBehaviour;
 import org.apache.qpid.exchange.ExchangeDefaults;
 import org.apache.qpid.url.AMQBindingURL;
+import org.apache.qpid.url.BindingURL;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -190,6 +193,67 @@ public class DestinationURLTest extends TestCase
         assertTrue(dest.getQueueName().equals("test:testQueueD"));
     }
 
+    public void testRejectBehaviourPresent() throws URISyntaxException
+    {
+        String url = "exchangeClass://exchangeName/Destination/Queue?rejectbehaviour='server'";
+
+        AMQBindingURL burl = new AMQBindingURL(url);
+
+        assertTrue(url.equals(burl.toString()));
+        assertTrue(burl.getExchangeClass().equals("exchangeClass"));
+        assertTrue(burl.getExchangeName().equals("exchangeName"));
+        assertTrue(burl.getDestinationName().equals("Destination"));
+        assertTrue(burl.getQueueName().equals("Queue"));
+
+        //check that the MaxDeliveryCount property has the right value
+        assertEquals("server",burl.getOption(BindingURL.OPTION_REJECT_BEHAVIOUR));
+
+        //check that the MaxDeliveryCount value is correctly returned from an AMQDestination
+        class MyTestAMQDestination extends AMQDestination
+        {
+            public MyTestAMQDestination(BindingURL url)
+            {
+                super(url);
+            }
+            public boolean isNameRequired()
+            {
+                return false;
+            }
+        };
+
+        AMQDestination dest = new MyTestAMQDestination(burl);
+        assertEquals("Reject behaviour is unexpected", RejectBehaviour.SERVER, dest.getRejectBehaviour());
+    }
+
+    public void testRejectBehaviourNotPresent() throws URISyntaxException
+    {
+        String url = "exchangeClass://exchangeName/Destination/Queue";
+
+        AMQBindingURL burl = new AMQBindingURL(url);
+
+        assertTrue(url.equals(burl.toString()));
+
+        assertTrue(burl.getExchangeClass().equals("exchangeClass"));
+        assertTrue(burl.getExchangeName().equals("exchangeName"));
+        assertTrue(burl.getDestinationName().equals("Destination"));
+        assertTrue(burl.getQueueName().equals("Queue"));
+
+        class MyTestAMQDestination extends AMQDestination
+        {
+            public MyTestAMQDestination(BindingURL url)
+            {
+                super(url);
+            }
+            public boolean isNameRequired()
+            {
+                return false;
+            }
+        };
+
+        AMQDestination dest = new MyTestAMQDestination(burl);
+        assertNull("Reject behaviour is unexpected", dest.getRejectBehaviour());
+    }
+
     public static junit.framework.Test suite()
     {
         return new junit.framework.TestSuite(DestinationURLTest.class);