diff options
author | Martin Ritchie <ritchiem@apache.org> | 2008-01-09 17:39:54 +0000 |
---|---|---|
committer | Martin Ritchie <ritchiem@apache.org> | 2008-01-09 17:39:54 +0000 |
commit | d67ac681fcea5188555f0b81c25f6b6edf97b77e (patch) | |
tree | fc9f01b8161ef65bbba3d02b2b0c67b2b14a1a96 | |
parent | 4dfca685eeeed97e9d36f1d5228d6579ea3b24bb (diff) | |
download | qpid-python-d67ac681fcea5188555f0b81c25f6b6edf97b77e.tar.gz |
Qpid-690 : Provide configurable delay between re-connecion attempts.
git-svn-id: https://svn.apache.org/repos/asf/incubator/qpid/branches/M2@610479 13f79535-47bb-0310-9956-ffa450edef68
4 files changed, 190 insertions, 10 deletions
diff --git a/java/client/src/main/java/org/apache/qpid/jms/BrokerDetails.java b/java/client/src/main/java/org/apache/qpid/jms/BrokerDetails.java index 91f7710025..30356f8910 100644 --- a/java/client/src/main/java/org/apache/qpid/jms/BrokerDetails.java +++ b/java/client/src/main/java/org/apache/qpid/jms/BrokerDetails.java @@ -31,6 +31,7 @@ public interface BrokerDetails */ public static final String OPTIONS_RETRY = "retries"; public static final String OPTIONS_CONNECT_TIMEOUT = "connecttimeout"; + public static final String OPTIONS_CONNECT_DELAY = "connectdelay"; public static final int DEFAULT_PORT = 5672; public static final String TCP = "tcp"; @@ -63,9 +64,9 @@ public interface BrokerDetails long getTimeout(); void setTimeout(long timeout); - + SSLConfiguration getSSLConfiguration(); - + void setSSLConfiguration(SSLConfiguration sslConfiguration); String toString(); diff --git a/java/client/src/main/java/org/apache/qpid/jms/failover/FailoverRoundRobinServers.java b/java/client/src/main/java/org/apache/qpid/jms/failover/FailoverRoundRobinServers.java index 4e0d0b79b5..b0756b220d 100644 --- a/java/client/src/main/java/org/apache/qpid/jms/failover/FailoverRoundRobinServers.java +++ b/java/client/src/main/java/org/apache/qpid/jms/failover/FailoverRoundRobinServers.java @@ -128,6 +128,8 @@ public class FailoverRoundRobinServers implements FailoverMethod public BrokerDetails getNextBrokerDetails() { + boolean doDelay = false; + if (_currentBrokerIndex == (_connectionDetails.getBrokerCount() - 1)) { if (_currentServerRetry < _serverRetries) @@ -143,6 +145,7 @@ public class FailoverRoundRobinServers implements FailoverMethod else { _logger.info("Retrying " + _connectionDetails.getBrokerDetails(_currentBrokerIndex)); + doDelay=true; } _currentServerRetry++; @@ -175,6 +178,7 @@ public class FailoverRoundRobinServers implements FailoverMethod else { _logger.info("Retrying " + _connectionDetails.getBrokerDetails(_currentBrokerIndex)); + doDelay=true; } _currentServerRetry++; @@ -189,7 +193,28 @@ public class FailoverRoundRobinServers implements FailoverMethod } } - return _connectionDetails.getBrokerDetails(_currentBrokerIndex); + BrokerDetails broker = _connectionDetails.getBrokerDetails(_currentBrokerIndex); + + String delayStr = broker.getOption(BrokerDetails.OPTIONS_CONNECT_DELAY); + if (delayStr != null && doDelay) + { + Long delay = Long.parseLong(delayStr); + _logger.info("Delay between connect retries:" + delay); + try + { + Thread.sleep(delay); + } + catch (InterruptedException ie) + { + return null; + } + } + else + { + _logger.info("No delay between connect retries, use tcp://host:port?connectdelay='value' to enable."); + } + + return broker; } public void setBroker(BrokerDetails broker) diff --git a/java/client/src/main/java/org/apache/qpid/jms/failover/FailoverSingleServer.java b/java/client/src/main/java/org/apache/qpid/jms/failover/FailoverSingleServer.java index 68e6d25be0..fcb4dee04c 100644 --- a/java/client/src/main/java/org/apache/qpid/jms/failover/FailoverSingleServer.java +++ b/java/client/src/main/java/org/apache/qpid/jms/failover/FailoverSingleServer.java @@ -22,9 +22,13 @@ package org.apache.qpid.jms.failover; import org.apache.qpid.jms.BrokerDetails; import org.apache.qpid.jms.ConnectionURL; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; public class FailoverSingleServer implements FailoverMethod { + private static final Logger _logger = LoggerFactory.getLogger(FailoverSingleServer.class); + /** The default number of times to rety a conection to this server */ public static final int DEFAULT_SERVER_RETRIES = 1; @@ -78,7 +82,7 @@ public class FailoverSingleServer implements FailoverMethod public BrokerDetails getCurrentBrokerDetails() { - return _brokerDetail; + return _brokerDetail; } public BrokerDetails getNextBrokerDetails() @@ -91,11 +95,29 @@ public class FailoverSingleServer implements FailoverMethod { if (_currentRetries < _retries) { - _currentRetries ++; + _currentRetries++; } + } + - return _brokerDetail; + String delayStr = _brokerDetail.getOption(BrokerDetails.OPTIONS_CONNECT_DELAY); + if (delayStr != null && _currentRetries != 1) + { + Long delay = Long.parseLong(delayStr); + _logger.info("Delay between connect retries:" + delay); + try + { + + Thread.sleep(delay); + } + catch (InterruptedException ie) + { + _logger.info("No delay between connect retries, use tcp://host:port?connectdelay='value' to enable."); + return null; + } } + + return _brokerDetail; } public void setBroker(BrokerDetails broker) @@ -138,10 +160,10 @@ public class FailoverSingleServer implements FailoverMethod public String toString() { - return "SingleServer:\n"+ - "Max Retries:"+_retries+ - "\nCurrent Retry:"+_currentRetries+ - "\n"+_brokerDetail+"\n"; + return "SingleServer:\n" + + "Max Retries:" + _retries + + "\nCurrent Retry:" + _currentRetries + + "\n" + _brokerDetail + "\n"; } } diff --git a/java/systests/src/main/java/org/apache/qpid/server/failover/FailoverMethodTest.java b/java/systests/src/main/java/org/apache/qpid/server/failover/FailoverMethodTest.java new file mode 100644 index 0000000000..b346dceef9 --- /dev/null +++ b/java/systests/src/main/java/org/apache/qpid/server/failover/FailoverMethodTest.java @@ -0,0 +1,132 @@ +/* + * + * 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.server.failover; + +import junit.framework.TestCase; +import org.apache.qpid.AMQDisconnectedException; +import org.apache.qpid.AMQException; +import org.apache.qpid.client.AMQConnection; +import org.apache.qpid.client.AMQConnectionURL; +import org.apache.qpid.client.transport.TransportConnection; +import org.apache.qpid.client.vmbroker.AMQVMBrokerCreationException; +import org.apache.qpid.url.URLSyntaxException; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import javax.jms.ExceptionListener; +import javax.jms.JMSException; +import java.util.concurrent.CountDownLatch; + +public class FailoverMethodTest extends TestCase implements ExceptionListener +{ + private CountDownLatch _failoverComplete = new CountDownLatch(1); + + public void setUp() throws AMQVMBrokerCreationException + { + TransportConnection.createVMBroker(1); + } + + public void tearDown() throws AMQVMBrokerCreationException + { + TransportConnection.killAllVMBrokers(); + } + + /** + * Test that the round robin method has the correct delays. + * The first connection to vm://:1 will work but the localhost connection should fail but the duration it takes + * to report the failure is what is being tested. + * + * @throws URLSyntaxException + * @throws InterruptedException + * @throws JMSException + */ + public void testFailoverRoundRobinDelay() throws URLSyntaxException, InterruptedException, JMSException + { + String connectionString = "amqp://guest:guest@/test?brokerlist='vm://:1;tcp://localhost:5670?connectdelay='2000',retries='3''"; + + AMQConnectionURL url = new AMQConnectionURL(connectionString); + + try + { + long start = System.currentTimeMillis(); + AMQConnection connection = new AMQConnection(url, null); + + connection.setExceptionListener(this); + + TransportConnection.killAllVMBrokers(); + + _failoverComplete.await(); + + long end = System.currentTimeMillis(); + + //Failover should take less that 10 seconds. + // This is calculated by vm://:1 two retries left after initial connection ( 4s) + // localhost get three retries so (6s) so 10s in total for connection dropping + assertTrue("Failover took over 10 seconds:"+(end - start), (end - start) > 10000); + + } + catch (AMQException e) + { + fail(e.getMessage()); + } + } + + public void testFailoverSingleDelay() throws URLSyntaxException, AMQVMBrokerCreationException, InterruptedException, JMSException + { + String connectionString = "amqp://guest:guest@/test?brokerlist='vm://:1?connectdelay='2000',retries='3''"; + + AMQConnectionURL url = new AMQConnectionURL(connectionString); + + try + { + long start = System.currentTimeMillis(); + AMQConnection connection = new AMQConnection(url, null); + + connection.setExceptionListener(this); + + TransportConnection.killAllVMBrokers(); + + _failoverComplete.await(); + + long end = System.currentTimeMillis(); + + //Failover should take less that 10 seconds. + // This is calculated by vm://:1 two retries left after initial connection + // so 4s in total for connection dropping + + assertTrue("Failover took over 4 seconds", (end - start) > 4000); + + } + catch (AMQException e) + { + fail(e.getMessage()); + } + } + + + public void onException(JMSException e) + { + if (e.getLinkedException() instanceof AMQDisconnectedException) + { + _failoverComplete.countDown(); + } + } +} |