From fd6e3bf54c79e8e050472838a9a68eb61b08ae13 Mon Sep 17 00:00:00 2001 From: Robert Gemmell Date: Mon, 16 May 2011 19:44:45 +0000 Subject: QPID-3220: Specifying connection url option failover='singlebroker' causes the wrong failover policy to be used. Wrote new unit test case and resolved the defect. Applied patch from Keith Wall git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk@1103857 13f79535-47bb-0310-9956-ffa450edef68 --- .../java/org/apache/qpid/jms/FailoverPolicy.java | 5 +- .../qpid/jms/failover/FailoverExchangeMethod.java | 6 +- .../org/apache/qpid/jms/FailoverPolicyTest.java | 338 +++++++++++++++++++++ 3 files changed, 343 insertions(+), 6 deletions(-) create mode 100644 qpid/java/client/src/test/java/org/apache/qpid/jms/FailoverPolicyTest.java diff --git a/qpid/java/client/src/main/java/org/apache/qpid/jms/FailoverPolicy.java b/qpid/java/client/src/main/java/org/apache/qpid/jms/FailoverPolicy.java index 7cdcd32306..56abf03c81 100644 --- a/qpid/java/client/src/main/java/org/apache/qpid/jms/FailoverPolicy.java +++ b/qpid/java/client/src/main/java/org/apache/qpid/jms/FailoverPolicy.java @@ -20,7 +20,6 @@ */ package org.apache.qpid.jms; -import org.apache.qpid.client.AMQConnection; import org.apache.qpid.jms.failover.FailoverExchangeMethod; import org.apache.qpid.jms.failover.FailoverMethod; import org.apache.qpid.jms.failover.FailoverRoundRobinServers; @@ -51,7 +50,7 @@ public class FailoverPolicy private long _lastMethodTime; private long _lastFailTime; - public FailoverPolicy(ConnectionURL connectionDetails, AMQConnection conn) + public FailoverPolicy(ConnectionURL connectionDetails, Connection conn) { FailoverMethod method; @@ -83,7 +82,7 @@ public class FailoverPolicy */ if (failoverMethod.equals(FailoverMethod.SINGLE_BROKER)) { - method = new FailoverRoundRobinServers(connectionDetails); + method = new FailoverSingleServer(connectionDetails); } else { diff --git a/qpid/java/client/src/main/java/org/apache/qpid/jms/failover/FailoverExchangeMethod.java b/qpid/java/client/src/main/java/org/apache/qpid/jms/failover/FailoverExchangeMethod.java index 9e6000c472..ef30f2adbc 100644 --- a/qpid/java/client/src/main/java/org/apache/qpid/jms/failover/FailoverExchangeMethod.java +++ b/qpid/java/client/src/main/java/org/apache/qpid/jms/failover/FailoverExchangeMethod.java @@ -32,9 +32,9 @@ import javax.jms.Session; import org.apache.qpid.client.AMQAnyDestination; import org.apache.qpid.client.AMQBrokerDetails; -import org.apache.qpid.client.AMQConnection; import org.apache.qpid.framing.AMQShortString; import org.apache.qpid.jms.BrokerDetails; +import org.apache.qpid.jms.Connection; import org.apache.qpid.jms.ConnectionURL; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -58,7 +58,7 @@ public class FailoverExchangeMethod implements FailoverMethod, MessageListener private static final Logger _logger = LoggerFactory.getLogger(FailoverExchangeMethod.class); /** This is not safe to use until attainConnection is called */ - private AMQConnection _conn; + private Connection _conn; /** Protects the broker list when modifications happens */ private Object _brokerListLock = new Object(); @@ -80,7 +80,7 @@ public class FailoverExchangeMethod implements FailoverMethod, MessageListener /** Denotes the number of failed attempts **/ private int _failedAttemps = 0; - public FailoverExchangeMethod(ConnectionURL connectionDetails, AMQConnection conn) + public FailoverExchangeMethod(ConnectionURL connectionDetails, Connection conn) { _connectionDetails = connectionDetails; _originalBrokerDetail = _connectionDetails.getBrokerDetails(0); diff --git a/qpid/java/client/src/test/java/org/apache/qpid/jms/FailoverPolicyTest.java b/qpid/java/client/src/test/java/org/apache/qpid/jms/FailoverPolicyTest.java new file mode 100644 index 0000000000..438995aedc --- /dev/null +++ b/qpid/java/client/src/test/java/org/apache/qpid/jms/FailoverPolicyTest.java @@ -0,0 +1,338 @@ +/* + * + * 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.jms; + +import javax.jms.ConnectionConsumer; +import javax.jms.ConnectionMetaData; +import javax.jms.Destination; +import javax.jms.ExceptionListener; +import javax.jms.JMSException; +import javax.jms.ServerSessionPool; +import javax.jms.Topic; + +import org.apache.qpid.client.AMQConnectionURL; +import org.apache.qpid.jms.failover.FailoverExchangeMethod; +import org.apache.qpid.jms.failover.FailoverMethod; +import org.apache.qpid.jms.failover.FailoverRoundRobinServers; +import org.apache.qpid.jms.failover.FailoverSingleServer; +import org.apache.qpid.jms.failover.NoFailover; + +import junit.framework.TestCase; + +/** + * Tests the ability of FailoverPolicy to instantiate the correct FailoverMethod. + * + * This test presently does not test {@link FailoverPolicy#FailoverPolicy(FailoverMethod) or + * {@link FailoverPolicy#addMethod(FailoverMethod)} as it appears that this functionality + * is no longer in use. + * + */ +public class FailoverPolicyTest extends TestCase +{ + private FailoverPolicy _failoverPolicy = null; // class under test + private String _url; + private Connection _connection = null; + private ConnectionURL _connectionUrl = null; + + /** + * Tests single server method is selected for a brokerlist with one broker when + * the failover option is not specified. + */ + public void testBrokerListWithOneBrokerDefaultsToSingleServerPolicy() throws Exception + { + _url = "amqp://user:pass@clientid/test?brokerlist='tcp://localhost:5672'"; + _connectionUrl = new AMQConnectionURL(_url); + _connection = createStubConnection(); + + _failoverPolicy = new FailoverPolicy(_connectionUrl, _connection); + + assertTrue("Unexpected failover method", _failoverPolicy.getCurrentMethod() instanceof FailoverSingleServer); + } + + /** + * Tests round robin method is selected for a brokerlist with two brokers when + * the failover option is not specified. + */ + public void testBrokerListWithTwoBrokersDefaultsToRoundRobinPolicy() throws Exception + { + _url = "amqp://user:pass@clientid/test?brokerlist='tcp://localhost:5672;tcp://localhost:5673'"; + _connectionUrl = new AMQConnectionURL(_url); + _connection = createStubConnection(); + + _failoverPolicy = new FailoverPolicy(_connectionUrl, _connection); + + assertTrue("Unexpected failover method", _failoverPolicy.getCurrentMethod() instanceof FailoverRoundRobinServers); + } + + /** + * Tests single server method is selected for a brokerlist with one broker when + * the failover option passed as 'singlebroker'. + */ + public void testExplictFailoverOptionSingleBroker() throws Exception + { + _url = "amqp://user:pass@clientid/test?brokerlist='tcp://localhost:5672'&failover='singlebroker'"; + _connectionUrl = new AMQConnectionURL(_url); + _connection = createStubConnection(); + + _failoverPolicy = new FailoverPolicy(_connectionUrl, _connection); + + assertTrue("Unexpected failover method", _failoverPolicy.getCurrentMethod() instanceof FailoverSingleServer); + } + + /** + * Tests round robin method is selected for a brokerlist with two brokers when + * the failover option passed as 'roundrobin'. + */ + public void testExplictFailoverOptionRoundrobin() throws Exception + { + _url = "amqp://user:pass@clientid/test?brokerlist='tcp://localhost:5672;tcp://localhost:5673'&failover='roundrobin'"; + _connectionUrl = new AMQConnectionURL(_url); + _connection = createStubConnection(); + + _failoverPolicy = new FailoverPolicy(_connectionUrl, _connection); + + assertTrue("Unexpected failover method", _failoverPolicy.getCurrentMethod() instanceof FailoverRoundRobinServers); + } + + /** + * Tests no failover method is selected for a brokerlist with one broker when + * the failover option passed as 'nofailover'. + */ + public void testExplictFailoverOptionNofailover() throws Exception + { + _url = "amqp://user:pass@clientid/test?brokerlist='tcp://localhost:5672'&failover='nofailover'"; + _connectionUrl = new AMQConnectionURL(_url); + _connection = createStubConnection(); + + _failoverPolicy = new FailoverPolicy(_connectionUrl, _connection); + + assertTrue("Unexpected failover method", _failoverPolicy.getCurrentMethod() instanceof NoFailover); + } + + /** + * Tests failover exchange method is selected for a brokerlist with one broker when + * the failover option passed as 'failover_exchange'. + */ + public void testExplictFailoverOptionFailoverExchange() throws Exception + { + _url = "amqp://user:pass@clientid/test?brokerlist='tcp://localhost:5672'&failover='failover_exchange'"; + _connectionUrl = new AMQConnectionURL(_url); + _connection = createStubConnection(); + + _failoverPolicy = new FailoverPolicy(_connectionUrl, _connection); + + assertTrue("Unexpected failover method", _failoverPolicy.getCurrentMethod() instanceof FailoverExchangeMethod); + } + + /** + * Tests that a custom method can be selected for a brokerlist with one brokers when + * the failover option passed as a qualified class-name. + */ + public void testExplictFailoverOptionDynamicallyLoadedFailoverMethod() throws Exception + { + _url = "amqp://user:pass@clientid/test?brokerlist='tcp://localhost:5672'&failover='org.apache.qpid.jms.FailoverPolicyTest$MyFailoverMethod'"; + _connectionUrl = new AMQConnectionURL(_url); + _connection = createStubConnection(); + + _failoverPolicy = new FailoverPolicy(_connectionUrl, _connection); + + assertTrue("Unexpected failover method", _failoverPolicy.getCurrentMethod() instanceof MyFailoverMethod); + } + + /** + * Tests that an unknown method caused an exception. + */ + public void testUnknownFailoverMethod() throws Exception + { + _url = "amqp://user:pass@clientid/test?brokerlist='tcp://localhost:5672'&failover='unknown'"; + _connectionUrl = new AMQConnectionURL(_url); + _connection = createStubConnection(); + + try + { + new FailoverPolicy(_connectionUrl, _connection); + fail("Exception not thrown"); + } + catch(IllegalArgumentException iae) + { + // PASS + } + } + + private Connection createStubConnection() + { + return new Connection() + { + + @Override + public Session createSession(boolean transacted, + int acknowledgeMode, int prefetch) throws JMSException + { + return null; + } + + @Override + public Session createSession(boolean transacted, + int acknowledgeMode, int prefetchHigh, int prefetchLow) + throws JMSException + { + return null; + } + + @Override + public ConnectionListener getConnectionListener() + { + return null; + } + + @Override + public long getMaximumChannelCount() throws JMSException + { + return 0; + } + + @Override + public void setConnectionListener(ConnectionListener listener) + { + } + + @Override + public void close() throws JMSException + { + } + + @Override + public ConnectionConsumer createConnectionConsumer( + Destination arg0, String arg1, ServerSessionPool arg2, + int arg3) throws JMSException + { + return null; + } + + @Override + public ConnectionConsumer createDurableConnectionConsumer( + Topic arg0, String arg1, String arg2, + ServerSessionPool arg3, int arg4) throws JMSException + { + return null; + } + + @Override + public javax.jms.Session createSession(boolean arg0, int arg1) + throws JMSException + { + return null; + } + + @Override + public String getClientID() throws JMSException + { + return null; + } + + @Override + public ExceptionListener getExceptionListener() throws JMSException + { + return null; + } + + @Override + public ConnectionMetaData getMetaData() throws JMSException + { + return null; + } + + @Override + public void setClientID(String arg0) throws JMSException + { + } + + @Override + public void setExceptionListener(ExceptionListener arg0) + throws JMSException + { + } + + @Override + public void start() throws JMSException + { + } + + @Override + public void stop() throws JMSException + { + } + }; + } + + // Class used to test the ability of FailoverPolicy to load an implementation. + static class MyFailoverMethod implements FailoverMethod + { + public MyFailoverMethod(ConnectionURL connectionDetails) + { + } + + @Override + public void attainedConnection() + { + } + + @Override + public boolean failoverAllowed() + { + return false; + } + + @Override + public BrokerDetails getCurrentBrokerDetails() + { + return null; + } + + @Override + public BrokerDetails getNextBrokerDetails() + { + return null; + } + + @Override + public String methodName() + { + return null; + } + + @Override + public void reset() + { + } + + @Override + public void setBroker(BrokerDetails broker) + { + } + + @Override + public void setRetries(int maxRetries) + { + } + } + +} -- cgit v1.2.1