You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by ta...@apache.org on 2016/03/18 20:42:25 UTC
qpid-jms git commit: QPIDJMS-157 Add support inf the FailoverProvider
for the sendTimeout connection option and the requestTimeout connection
option.
Repository: qpid-jms
Updated Branches:
refs/heads/master b5776684a -> 4e963e442
QPIDJMS-157 Add support inf the FailoverProvider for the sendTimeout
connection option and the requestTimeout connection option.
Project: http://git-wip-us.apache.org/repos/asf/qpid-jms/repo
Commit: http://git-wip-us.apache.org/repos/asf/qpid-jms/commit/4e963e44
Tree: http://git-wip-us.apache.org/repos/asf/qpid-jms/tree/4e963e44
Diff: http://git-wip-us.apache.org/repos/asf/qpid-jms/diff/4e963e44
Branch: refs/heads/master
Commit: 4e963e4427308b3c3932041427c0c243e82ea556
Parents: b577668
Author: Timothy Bish <ta...@gmail.com>
Authored: Fri Mar 18 15:42:09 2016 -0400
Committer: Timothy Bish <ta...@gmail.com>
Committed: Fri Mar 18 15:42:09 2016 -0400
----------------------------------------------------------------------
.../qpid/jms/JmsOperationTimedOutException.java | 36 ++++
.../qpid/jms/JmsSendTimedOutException.java | 52 +++++
.../qpid/jms/provider/amqp/AmqpProvider.java | 3 +-
.../jms/provider/failover/FailoverProvider.java | 130 ++++++++++--
.../failover/FailoverIntegrationTest.java | 208 ++++++++++++++++++-
.../provider/failover/FailoverProviderTest.java | 1 -
6 files changed, 398 insertions(+), 32 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/qpid-jms/blob/4e963e44/qpid-jms-client/src/main/java/org/apache/qpid/jms/JmsOperationTimedOutException.java
----------------------------------------------------------------------
diff --git a/qpid-jms-client/src/main/java/org/apache/qpid/jms/JmsOperationTimedOutException.java b/qpid-jms-client/src/main/java/org/apache/qpid/jms/JmsOperationTimedOutException.java
new file mode 100644
index 0000000..1a9348f
--- /dev/null
+++ b/qpid-jms-client/src/main/java/org/apache/qpid/jms/JmsOperationTimedOutException.java
@@ -0,0 +1,36 @@
+/*
+ * 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.JMSException;
+
+/**
+ * Exception that indicates a blocking operation timed out while waiting
+ * for the remote to acknowledge or process it.
+ */
+public class JmsOperationTimedOutException extends JMSException {
+
+ private static final long serialVersionUID = -2509921066407708297L;
+
+ public JmsOperationTimedOutException(String reason) {
+ super(reason);
+ }
+
+ public JmsOperationTimedOutException(String reason, String errorCode) {
+ super(reason, errorCode);
+ }
+}
http://git-wip-us.apache.org/repos/asf/qpid-jms/blob/4e963e44/qpid-jms-client/src/main/java/org/apache/qpid/jms/JmsSendTimedOutException.java
----------------------------------------------------------------------
diff --git a/qpid-jms-client/src/main/java/org/apache/qpid/jms/JmsSendTimedOutException.java b/qpid-jms-client/src/main/java/org/apache/qpid/jms/JmsSendTimedOutException.java
new file mode 100644
index 0000000..36a305b
--- /dev/null
+++ b/qpid-jms-client/src/main/java/org/apache/qpid/jms/JmsSendTimedOutException.java
@@ -0,0 +1,52 @@
+/*
+ * 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.Message;
+
+import org.apache.qpid.jms.message.JmsMessage;
+
+/**
+ * Exception thrown when a blocking send call times out waiting to be sent.
+ */
+public class JmsSendTimedOutException extends JmsOperationTimedOutException {
+
+ private static final long serialVersionUID = -5404637534865352615L;
+
+ private final JmsMessage unsentMessage;
+
+ public JmsSendTimedOutException(String reason) {
+ this(reason, null, null);
+ }
+
+ public JmsSendTimedOutException(String reason, String errorCode) {
+ this(reason, errorCode, null);
+ }
+
+ public JmsSendTimedOutException(String reason, JmsMessage unsentMessage) {
+ this(reason, null, unsentMessage);
+ }
+
+ public JmsSendTimedOutException(String reason, String errorCode, JmsMessage unsentMessage) {
+ super(reason, errorCode);
+ this.unsentMessage = unsentMessage;
+ }
+
+ public Message getUnsentMessage() {
+ return unsentMessage;
+ }
+}
http://git-wip-us.apache.org/repos/asf/qpid-jms/blob/4e963e44/qpid-jms-client/src/main/java/org/apache/qpid/jms/provider/amqp/AmqpProvider.java
----------------------------------------------------------------------
diff --git a/qpid-jms-client/src/main/java/org/apache/qpid/jms/provider/amqp/AmqpProvider.java b/qpid-jms-client/src/main/java/org/apache/qpid/jms/provider/amqp/AmqpProvider.java
index 4980bfb..4cb1028 100644
--- a/qpid-jms-client/src/main/java/org/apache/qpid/jms/provider/amqp/AmqpProvider.java
+++ b/qpid-jms-client/src/main/java/org/apache/qpid/jms/provider/amqp/AmqpProvider.java
@@ -226,8 +226,7 @@ public class AmqpProvider implements Provider, TransportListener , AmqpResourceP
request.sync(closeTimeout, TimeUnit.MILLISECONDS);
}
} catch (IOException e) {
- LOG.warn("Error caught while closing Provider: ", e.getMessage());
- // TODO: message can be / seemingly usually is empty
+ LOG.warn("Error caught while closing Provider: {}", e.getMessage() != null ? e.getMessage() : "<Unknown Error>");
} finally {
if (transport != null) {
try {
http://git-wip-us.apache.org/repos/asf/qpid-jms/blob/4e963e44/qpid-jms-client/src/main/java/org/apache/qpid/jms/provider/failover/FailoverProvider.java
----------------------------------------------------------------------
diff --git a/qpid-jms-client/src/main/java/org/apache/qpid/jms/provider/failover/FailoverProvider.java b/qpid-jms-client/src/main/java/org/apache/qpid/jms/provider/failover/FailoverProvider.java
index a8e67bf..6399080 100644
--- a/qpid-jms-client/src/main/java/org/apache/qpid/jms/provider/failover/FailoverProvider.java
+++ b/qpid-jms-client/src/main/java/org/apache/qpid/jms/provider/failover/FailoverProvider.java
@@ -20,12 +20,13 @@ import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
-import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
@@ -35,6 +36,8 @@ import java.util.concurrent.atomic.AtomicReference;
import javax.jms.JMSException;
import javax.jms.JMSSecurityException;
+import org.apache.qpid.jms.JmsOperationTimedOutException;
+import org.apache.qpid.jms.JmsSendTimedOutException;
import org.apache.qpid.jms.message.JmsInboundMessageDispatch;
import org.apache.qpid.jms.message.JmsMessageFactory;
import org.apache.qpid.jms.message.JmsOutboundMessageDispatch;
@@ -68,7 +71,10 @@ public class FailoverProvider extends DefaultProviderListener implements Provide
private static final Logger LOG = LoggerFactory.getLogger(FailoverProvider.class);
public static final int UNLIMITED = -1;
+
private static final int UNDEFINED = -1;
+ private static final int DISABLED = 0;
+ private static final int MINIMUM_TIMEOUT = 1000;
public static final int DEFAULT_MAX_RECONNECT_ATTEMPTS = UNLIMITED;
public static final int DEFAULT_STARTUP_MAX_RECONNECT_ATTEMPTS = UNDEFINED;
@@ -82,8 +88,9 @@ public class FailoverProvider extends DefaultProviderListener implements Provide
private ProviderListener listener;
private Provider provider;
private final FailoverUriPool uris;
+ private ScheduledFuture<?> requestTimeoutTask;
- private final ExecutorService serializer;
+ private final ScheduledExecutorService serializer;
private final ScheduledExecutorService connectionHub;
private final AtomicBoolean closed = new AtomicBoolean();
private final AtomicBoolean failed = new AtomicBoolean();
@@ -102,9 +109,8 @@ public class FailoverProvider extends DefaultProviderListener implements Provide
private URI connectedURI;
// Timeout values configured via JmsConnectionInfo
- private long connectTimeout = JmsConnectionInfo.DEFAULT_CONNECT_TIMEOUT;
private long closeTimeout = JmsConnectionInfo.DEFAULT_CLOSE_TIMEOUT;
- private long sendTimeout = JmsConnectionInfo.DEFAULT_SEND_TIMEOUT;
+ private long sendTimeout = JmsConnectionInfo.DEFAULT_SEND_TIMEOUT;
private long requestTimeout = JmsConnectionInfo.DEFAULT_REQUEST_TIMEOUT;
// Configuration values.
@@ -185,6 +191,10 @@ public class FailoverProvider extends DefaultProviderListener implements Provide
request.onFailure(error);
}
+ if (requestTimeoutTask != null) {
+ requestTimeoutTask.cancel(false);
+ }
+
if (provider != null) {
provider.close();
}
@@ -204,7 +214,7 @@ public class FailoverProvider extends DefaultProviderListener implements Provide
if (this.closeTimeout < 0) {
request.sync();
} else {
- request.sync(closeTimeout, TimeUnit.MILLISECONDS);
+ request.sync(Math.max(MINIMUM_TIMEOUT, closeTimeout), TimeUnit.MILLISECONDS);
}
} catch (IOException e) {
LOG.warn("Error caught while closing Provider: ", e.getMessage());
@@ -221,7 +231,8 @@ public class FailoverProvider extends DefaultProviderListener implements Provide
@Override
public void doTask() throws Exception {
JmsConnectionInfo connectionInfo = (JmsConnectionInfo) resource;
- connectTimeout = connectionInfo.getConnectTimeout();
+
+ // Collect the timeouts we will handle in this provider.
closeTimeout = connectionInfo.getCloseTimeout();
sendTimeout = connectionInfo.getSendTimeout();
requestTimeout = connectionInfo.getRequestTimeout();
@@ -235,7 +246,7 @@ public class FailoverProvider extends DefaultProviderListener implements Provide
}
};
} else {
- pending = new FailoverRequest(request) {
+ pending = new FailoverRequest(request, requestTimeout) {
@Override
public void doTask() throws Exception {
provider.create(resource, this);
@@ -267,7 +278,7 @@ public class FailoverProvider extends DefaultProviderListener implements Provide
@Override
public void start(final JmsResource resource, final AsyncResult request) throws IOException, JMSException {
checkClosed();
- final FailoverRequest pending = new FailoverRequest(request) {
+ final FailoverRequest pending = new FailoverRequest(request, requestTimeout) {
@Override
public void doTask() throws Exception {
provider.start(resource, this);
@@ -285,7 +296,7 @@ public class FailoverProvider extends DefaultProviderListener implements Provide
@Override
public void stop(final JmsResource resource, final AsyncResult request) throws IOException, JMSException {
checkClosed();
- final FailoverRequest pending = new FailoverRequest(request) {
+ final FailoverRequest pending = new FailoverRequest(request, requestTimeout) {
@Override
public void doTask() throws Exception {
provider.stop(resource, this);
@@ -303,7 +314,7 @@ public class FailoverProvider extends DefaultProviderListener implements Provide
@Override
public void destroy(final JmsResource resourceId, AsyncResult request) throws IOException, JMSException, UnsupportedOperationException {
checkClosed();
- final FailoverRequest pending = new FailoverRequest(request) {
+ final FailoverRequest pending = new FailoverRequest(request, requestTimeout) {
@Override
public void doTask() throws IOException, JMSException, UnsupportedOperationException {
if (resourceId instanceof JmsConnectionInfo) {
@@ -330,7 +341,7 @@ public class FailoverProvider extends DefaultProviderListener implements Provide
@Override
public void send(final JmsOutboundMessageDispatch envelope, AsyncResult request) throws IOException, JMSException {
checkClosed();
- final FailoverRequest pending = new FailoverRequest(request) {
+ final FailoverRequest pending = new FailoverRequest(request, sendTimeout) {
@Override
public void doTask() throws Exception {
provider.send(envelope, this);
@@ -340,6 +351,11 @@ public class FailoverProvider extends DefaultProviderListener implements Provide
public String toString() {
return "send -> " + envelope;
}
+
+ @Override
+ public JMSException createTimedOutException() {
+ return new JmsSendTimedOutException("Timed out waiting on " + this, envelope.getMessage());
+ }
};
serializer.execute(pending);
@@ -348,7 +364,7 @@ public class FailoverProvider extends DefaultProviderListener implements Provide
@Override
public void acknowledge(final JmsSessionId sessionId, final ACK_TYPE ackType, AsyncResult request) throws IOException, JMSException {
checkClosed();
- final FailoverRequest pending = new FailoverRequest(request) {
+ final FailoverRequest pending = new FailoverRequest(request, requestTimeout) {
@Override
public void doTask() throws Exception {
provider.acknowledge(sessionId, ackType, this);
@@ -372,7 +388,7 @@ public class FailoverProvider extends DefaultProviderListener implements Provide
@Override
public void acknowledge(final JmsInboundMessageDispatch envelope, final ACK_TYPE ackType, AsyncResult request) throws IOException, JMSException {
checkClosed();
- final FailoverRequest pending = new FailoverRequest(request) {
+ final FailoverRequest pending = new FailoverRequest(request, requestTimeout) {
@Override
public void doTask() throws Exception {
provider.acknowledge(envelope, ackType, this);
@@ -396,7 +412,7 @@ public class FailoverProvider extends DefaultProviderListener implements Provide
@Override
public void commit(final JmsTransactionInfo transactionInfo, AsyncResult request) throws IOException, JMSException, UnsupportedOperationException {
checkClosed();
- final FailoverRequest pending = new FailoverRequest(request) {
+ final FailoverRequest pending = new FailoverRequest(request, requestTimeout) {
@Override
public void doTask() throws Exception {
provider.commit(transactionInfo, this);
@@ -419,7 +435,7 @@ public class FailoverProvider extends DefaultProviderListener implements Provide
@Override
public void rollback(final JmsTransactionInfo transactionInfo, AsyncResult request) throws IOException, JMSException, UnsupportedOperationException {
checkClosed();
- final FailoverRequest pending = new FailoverRequest(request) {
+ final FailoverRequest pending = new FailoverRequest(request, requestTimeout) {
@Override
public void doTask() throws Exception {
provider.rollback(transactionInfo, this);
@@ -442,7 +458,7 @@ public class FailoverProvider extends DefaultProviderListener implements Provide
@Override
public void recover(final JmsSessionId sessionId, final AsyncResult request) throws IOException, UnsupportedOperationException {
checkClosed();
- final FailoverRequest pending = new FailoverRequest(request) {
+ final FailoverRequest pending = new FailoverRequest(request, requestTimeout) {
@Override
public void doTask() throws Exception {
provider.recover(sessionId, this);
@@ -465,7 +481,7 @@ public class FailoverProvider extends DefaultProviderListener implements Provide
@Override
public void unsubscribe(final String subscription, AsyncResult request) throws IOException, JMSException, UnsupportedOperationException {
checkClosed();
- final FailoverRequest pending = new FailoverRequest(request) {
+ final FailoverRequest pending = new FailoverRequest(request, requestTimeout) {
@Override
public void doTask() throws Exception {
provider.unsubscribe(subscription, this);
@@ -543,6 +559,17 @@ public class FailoverProvider extends DefaultProviderListener implements Provide
if (listener != null) {
listener.onConnectionInterrupted(failedURI);
}
+
+ // Start watching for request timeouts while we are offline, unless we already are.
+ if (requestTimeoutTask == null) {
+ long sweeperInterval = getRequestSweeperInterval();
+ if (sweeperInterval > 0) {
+ LOG.trace("Request timeout monitoring enabled: interval = {}ms", sweeperInterval);
+ requestTimeoutTask = serializer.scheduleWithFixedDelay(
+ new FailoverRequestSweeper(), sweeperInterval, sweeperInterval, TimeUnit.MILLISECONDS);
+ }
+ }
+
triggerReconnectionAttempt();
} else {
ProviderListener listener = this.listener;
@@ -595,6 +622,16 @@ public class FailoverProvider extends DefaultProviderListener implements Provide
reconnectAttempts = 0;
connectedURI = provider.getRemoteURI();
uris.connected();
+
+ // Cancel timeout processing since we are connected again. We waited until
+ // now for the case where we are continually getting bounced from otherwise
+ // live servers, we want the timeout to remain scheduled in that case so that
+ // it doesn't keep getting rescheduled and never actually time anything out.
+ if (requestTimeoutTask != null) {
+ requestTimeoutTask.cancel(false);
+ requestTimeoutTask = null;
+ }
+
} catch (Throwable error) {
handleProviderFailure(IOExceptionSupport.create(error));
}
@@ -911,10 +948,6 @@ public class FailoverProvider extends DefaultProviderListener implements Provide
this.useReconnectBackOff = useReconnectBackOff;
}
- public long getConnectTimeout() {
- return this.connectTimeout;
- }
-
public long getCloseTimeout() {
return this.closeTimeout;
}
@@ -937,7 +970,37 @@ public class FailoverProvider extends DefaultProviderListener implements Provide
(connectedURI == null ? "unconnected" : connectedURI.toString());
}
- //--------------- FailoverProvider Asynchronous Request --------------------//
+ //--------------- FailoverProvider Request Timeout Support ---------------//
+
+ protected final long getRequestSweeperInterval() {
+ long[] timeouts = new long[] { requestTimeout, sendTimeout };
+
+ Arrays.sort(timeouts);
+
+ for (long timeout : timeouts) {
+ if (timeout != JmsConnectionInfo.INFINITE) {
+ return Math.max(Math.max(1, timeout) / 3, MINIMUM_TIMEOUT);
+ }
+ }
+
+ return DISABLED;
+ }
+
+ protected final class FailoverRequestSweeper implements Runnable {
+
+ @Override
+ public void run() {
+ List<FailoverRequest> copied = new ArrayList<FailoverRequest>(requests.values());
+ for (FailoverRequest request : copied) {
+ if (request.isExpired()) {
+ LOG.trace("Task {} has timed out, sending failure notice.", request);
+ request.onFailure(request.createTimedOutException());
+ }
+ }
+ }
+ }
+
+ //--------------- FailoverProvider Asynchronous Request ------------------//
/**
* For all requests that are dispatched from the FailoverProvider to a connected
@@ -948,8 +1011,16 @@ public class FailoverProvider extends DefaultProviderListener implements Provide
protected final long id = requestId.incrementAndGet();
+ private final long requestStarted = System.nanoTime();
+ private final long requestTimeout;
+
public FailoverRequest(AsyncResult watcher) {
+ this(watcher, JmsConnectionInfo.INFINITE);
+ }
+
+ public FailoverRequest(AsyncResult watcher, long requestTimeout) {
super(watcher);
+ this.requestTimeout = requestTimeout;
LOG.trace("Created Failover Task: {} ({})", this, id);
}
@@ -1023,6 +1094,21 @@ public class FailoverProvider extends DefaultProviderListener implements Provide
return false;
}
+ /**
+ * @return true if the request has a configured expiration time.
+ */
+ public boolean isExpired() {
+ if (requestTimeout != JmsConnectionInfo.INFINITE) {
+ return (System.nanoTime() - requestStarted) > TimeUnit.MILLISECONDS.toNanos(requestTimeout);
+ } else {
+ return false;
+ }
+ }
+
+ public JMSException createTimedOutException() {
+ return new JmsOperationTimedOutException("Timed out waiting on " + this);
+ }
+
private void whenOffline(IOException error) {
if (failureWhenOffline()) {
requests.remove(id);
http://git-wip-us.apache.org/repos/asf/qpid-jms/blob/4e963e44/qpid-jms-client/src/test/java/org/apache/qpid/jms/provider/failover/FailoverIntegrationTest.java
----------------------------------------------------------------------
diff --git a/qpid-jms-client/src/test/java/org/apache/qpid/jms/provider/failover/FailoverIntegrationTest.java b/qpid-jms-client/src/test/java/org/apache/qpid/jms/provider/failover/FailoverIntegrationTest.java
index 3b79e46..6b6dcb5 100644
--- a/qpid-jms-client/src/test/java/org/apache/qpid/jms/provider/failover/FailoverIntegrationTest.java
+++ b/qpid-jms-client/src/test/java/org/apache/qpid/jms/provider/failover/FailoverIntegrationTest.java
@@ -48,7 +48,9 @@ import javax.jms.Topic;
import org.apache.qpid.jms.JmsConnection;
import org.apache.qpid.jms.JmsConnectionFactory;
import org.apache.qpid.jms.JmsDefaultConnectionListener;
+import org.apache.qpid.jms.JmsOperationTimedOutException;
import org.apache.qpid.jms.JmsPrefetchPolicy;
+import org.apache.qpid.jms.JmsSendTimedOutException;
import org.apache.qpid.jms.test.QpidJmsTestCase;
import org.apache.qpid.jms.test.testpeer.TestAmqpPeer;
import org.apache.qpid.jms.test.testpeer.basictypes.AmqpError;
@@ -384,7 +386,6 @@ public class FailoverIntegrationTest extends QpidJmsTestCase {
}
}
-
@Test(timeout = 30000)
public void testFailoverInitialReconnectDelayDoesNotApplyToInitialConnect() throws Exception {
try (TestAmqpPeer originalPeer = new TestAmqpPeer();) {
@@ -773,7 +774,7 @@ public class FailoverIntegrationTest extends QpidJmsTestCase {
@Test(timeout=20000)
public void testTxRecreatedAfterConnectionFailsOver() throws Exception {
try (TestAmqpPeer originalPeer = new TestAmqpPeer();
- TestAmqpPeer finalPeer = new TestAmqpPeer();) {
+ TestAmqpPeer finalPeer = new TestAmqpPeer();) {
final CountDownLatch originalConnected = new CountDownLatch(1);
final CountDownLatch finalConnected = new CountDownLatch(1);
@@ -848,22 +849,211 @@ public class FailoverIntegrationTest extends QpidJmsTestCase {
}
}
+ @Test(timeout = 20000)
+ public void testFailoverEnforcesRequestTimeoutSession() throws Exception {
+ try (TestAmqpPeer testPeer = new TestAmqpPeer()) {
+
+ final CountDownLatch connected = new CountDownLatch(1);
+ final CountDownLatch disconnected = new CountDownLatch(1);
+
+ // Create a peer to connect to so we can get to a state where we
+ // can try to send when offline.
+ final String peerURI = createPeerURI(testPeer);
+
+ LOG.info("Original peer is at: {}", peerURI);
+
+ // Connect to the test peer
+ testPeer.expectSaslAnonymousConnect();
+ testPeer.expectBegin();
+ testPeer.dropAfterLastHandler();
+
+ final JmsConnection connection = establishAnonymousConnecton(
+ "jms.requestTimeout=2000&failover.reconnectDelay=2000&failover.maxReconnectAttempts=60",
+ testPeer);
+ connection.addConnectionListener(new JmsDefaultConnectionListener() {
+
+ @Override
+ public void onConnectionInterrupted(URI remoteURI) {
+ LOG.info("Connection Interrupted: {}", remoteURI);
+ disconnected.countDown();
+ }
+
+ @Override
+ public void onConnectionEstablished(URI remoteURI) {
+ LOG.info("Connection Established: {}", remoteURI);
+ connected.countDown();
+ }
+ });
+ connection.start();
+
+ assertTrue("Should connect to peer", connected.await(5, TimeUnit.SECONDS));
+ assertTrue("Should lose connection to peer", disconnected.await(5, TimeUnit.SECONDS));
+
+ try {
+ connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ fail("Should have thrown an exception");
+ } catch (JmsOperationTimedOutException jmsEx) {
+ LOG.info("Caught timed out exception from send:", jmsEx);
+ } catch (Exception ex) {
+ fail("Should have caught a timed out exception");
+ }
+
+ connection.close();
+
+ testPeer.waitForAllHandlersToComplete(1000);
+ }
+ }
+
+ @Test(timeout = 20000)
+ public void testFailoverEnforcesSendTimeout() throws Exception {
+ try (TestAmqpPeer testPeer = new TestAmqpPeer()) {
+
+ final CountDownLatch connected = new CountDownLatch(1);
+ final CountDownLatch disconnected = new CountDownLatch(1);
+
+ // Create a peer to connect to so we can get to a state where we
+ // can try to send when offline.
+ final String peerURI = createPeerURI(testPeer);
+
+ LOG.info("Original peer is at: {}", peerURI);
+
+ // Connect to the test peer
+ testPeer.expectSaslAnonymousConnect();
+ testPeer.expectBegin();
+ testPeer.expectBegin();
+ testPeer.expectSenderAttach();
+ testPeer.dropAfterLastHandler();
+
+ final JmsConnection connection = establishAnonymousConnecton(
+ "jms.sendTimeout=2000&failover.reconnectDelay=2000&failover.maxReconnectAttempts=60",
+ testPeer);
+ connection.addConnectionListener(new JmsDefaultConnectionListener() {
+
+ @Override
+ public void onConnectionInterrupted(URI remoteURI) {
+ LOG.info("Connection Interrupted: {}", remoteURI);
+ disconnected.countDown();
+ }
+
+ @Override
+ public void onConnectionEstablished(URI remoteURI) {
+ LOG.info("Connection Established: {}", remoteURI);
+ connected.countDown();
+ }
+ });
+ connection.start();
+
+ assertTrue("Should connect to peer", connected.await(5, TimeUnit.SECONDS));
+
+ Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ Queue queue = session.createQueue("myQueue");
+ MessageProducer producer = session.createProducer(queue);
+
+ assertTrue("Should lose connection to peer", disconnected.await(5, TimeUnit.SECONDS));
+
+ try {
+ producer.send(session.createMessage());
+ fail("Should have thrown an exception");
+ } catch (JmsSendTimedOutException jmsEx) {
+ LOG.info("Caught timed out exception from send:", jmsEx);
+ } catch (Exception ex) {
+ fail("Should have caught a timed out exception");
+ }
+
+ connection.close();
+
+ testPeer.waitForAllHandlersToComplete(1000);
+ }
+ }
+
+ @Test(timeout = 20000)
+ public void testFailoverEnforcesRequestTimeoutCreateTenpDestination() throws Exception {
+ try (TestAmqpPeer testPeer = new TestAmqpPeer()) {
+
+ final CountDownLatch connected = new CountDownLatch(1);
+ final CountDownLatch disconnected = new CountDownLatch(1);
+
+ // Create a peer to connect to so we can get to a state where we
+ // can try to send when offline.
+ final String peerURI = createPeerURI(testPeer);
+
+ LOG.info("Original peer is at: {}", peerURI);
+
+ // Connect to the test peer
+ testPeer.expectSaslAnonymousConnect();
+ testPeer.expectBegin();
+ testPeer.expectBegin();
+ testPeer.dropAfterLastHandler();
+
+ final JmsConnection connection = establishAnonymousConnecton(
+ "jms.requestTimeout=2000&failover.reconnectDelay=2000&failover.maxReconnectAttempts=60",
+ testPeer);
+ connection.addConnectionListener(new JmsDefaultConnectionListener() {
+
+ @Override
+ public void onConnectionInterrupted(URI remoteURI) {
+ LOG.info("Connection Interrupted: {}", remoteURI);
+ disconnected.countDown();
+ }
+
+ @Override
+ public void onConnectionEstablished(URI remoteURI) {
+ LOG.info("Connection Established: {}", remoteURI);
+ connected.countDown();
+ }
+ });
+ connection.start();
+
+ assertTrue("Should connect to peer", connected.await(5, TimeUnit.SECONDS));
+
+ Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+
+ assertTrue("Should lose connection to peer", disconnected.await(5, TimeUnit.SECONDS));
+
+ try {
+ session.createTemporaryQueue();
+ fail("Should have thrown an exception");
+ } catch (JmsOperationTimedOutException jmsEx) {
+ LOG.info("Caught timed out exception from send:", jmsEx);
+ } catch (Exception ex) {
+ fail("Should have caught a timed out exception");
+ }
+
+ try {
+ session.createTemporaryTopic();
+ fail("Should have thrown an exception");
+ } catch (JmsOperationTimedOutException jmsEx) {
+ LOG.info("Caught timed out exception from send:", jmsEx);
+ } catch (Exception ex) {
+ fail("Should have caught a timed out exception");
+ }
+
+ connection.close();
+
+ testPeer.waitForAllHandlersToComplete(1000);
+ }
+ }
+
private JmsConnection establishAnonymousConnecton(TestAmqpPeer... peers) throws JMSException {
- return establishAnonymousConnecton(null, peers);
+ return establishAnonymousConnecton(null, null, peers);
}
private JmsConnection establishAnonymousConnecton(String failoverParams, TestAmqpPeer... peers) throws JMSException {
- if(peers.length == 0) {
+ return establishAnonymousConnecton(null, failoverParams, peers);
+ }
+
+ private JmsConnection establishAnonymousConnecton(String connectionParams, String failoverParams, TestAmqpPeer... peers) throws JMSException {
+ if (peers.length == 0) {
throw new IllegalArgumentException("No test peers were given, at least 1 required");
}
String remoteURI = "failover:(";
boolean first = true;
- for(TestAmqpPeer peer : peers) {
+ for (TestAmqpPeer peer : peers) {
if (!first) {
remoteURI += ",";
}
- remoteURI += createPeerURI(peer);
+ remoteURI += createPeerURI(peer, connectionParams);
first = false;
}
@@ -880,6 +1070,10 @@ public class FailoverIntegrationTest extends QpidJmsTestCase {
}
private String createPeerURI(TestAmqpPeer peer) {
- return "amqp://localhost:" + peer.getServerPort();
+ return createPeerURI(peer, null);
+ }
+
+ private String createPeerURI(TestAmqpPeer peer, String params) {
+ return "amqp://localhost:" + peer.getServerPort() + (params != null ? "?" + params : "");
}
}
http://git-wip-us.apache.org/repos/asf/qpid-jms/blob/4e963e44/qpid-jms-client/src/test/java/org/apache/qpid/jms/provider/failover/FailoverProviderTest.java
----------------------------------------------------------------------
diff --git a/qpid-jms-client/src/test/java/org/apache/qpid/jms/provider/failover/FailoverProviderTest.java b/qpid-jms-client/src/test/java/org/apache/qpid/jms/provider/failover/FailoverProviderTest.java
index 31f042b..cd46ad7 100644
--- a/qpid-jms-client/src/test/java/org/apache/qpid/jms/provider/failover/FailoverProviderTest.java
+++ b/qpid-jms-client/src/test/java/org/apache/qpid/jms/provider/failover/FailoverProviderTest.java
@@ -406,7 +406,6 @@ public class FailoverProviderTest extends FailoverProviderTestSupport {
provider.create(connectionInfo, request);
request.sync();
- assertEquals(CONNECT_TIMEOUT, provider.getConnectTimeout());
assertEquals(CLOSE_TIMEOUT, provider.getCloseTimeout());
assertEquals(SEND_TIMEOUT, provider.getSendTimeout());
assertEquals(REQUEST_TIMEOUT, provider.getRequestTimeout());
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org