diff options
Diffstat (limited to 'qpid/java/client/src/main/java/org/apache/qpid/client/AMQConnectionDelegate_0_10.java')
-rw-r--r-- | qpid/java/client/src/main/java/org/apache/qpid/client/AMQConnectionDelegate_0_10.java | 14 |
1 files changed, 10 insertions, 4 deletions
diff --git a/qpid/java/client/src/main/java/org/apache/qpid/client/AMQConnectionDelegate_0_10.java b/qpid/java/client/src/main/java/org/apache/qpid/client/AMQConnectionDelegate_0_10.java index 4b4417b6ef..b0bd8f8e97 100644 --- a/qpid/java/client/src/main/java/org/apache/qpid/client/AMQConnectionDelegate_0_10.java +++ b/qpid/java/client/src/main/java/org/apache/qpid/client/AMQConnectionDelegate_0_10.java @@ -26,6 +26,7 @@ import java.util.ArrayList; import java.util.HashMap; import java.util.List; import java.util.Map; +import java.util.concurrent.CountDownLatch; import javax.jms.ExceptionListener; import javax.jms.JMSException; @@ -211,15 +212,13 @@ public class AMQConnectionDelegate_0_10 implements AMQConnectionDelegate, Connec public void resubscribeSessions() throws JMSException, AMQException, FailoverException { List<AMQSession> sessions = new ArrayList<AMQSession>(_conn.getSessions().values()); - _logger.info(String.format("Resubscribing sessions = %s sessions.size=%s", sessions, sessions.size())); + _logger.info(String.format("Resubscribing sessions = %s sessions.size=%d", sessions, sessions.size())); for (AMQSession s : sessions) { - ((AMQSession_0_10) s)._qpidConnection = _qpidConnection; s.resubscribe(); } } - public void closeConnection(long timeout) throws JMSException, AMQException { try @@ -257,12 +256,14 @@ public class AMQConnectionDelegate_0_10 implements AMQConnectionDelegate, Connec ConnectionClose close = exc.getClose(); if (close == null) { + _conn.getProtocolHandler().setFailoverLatch(new CountDownLatch(1)); + try { if (_conn.firePreFailover(false) && _conn.attemptReconnection()) { _conn.failoverPrep(); - _qpidConnection.resume(); + _conn.resubscribeSessions(); _conn.fireFailoverComplete(); return; } @@ -271,6 +272,11 @@ public class AMQConnectionDelegate_0_10 implements AMQConnectionDelegate, Connec { _logger.error("error during failover", e); } + finally + { + _conn.getProtocolHandler().getFailoverLatch().countDown(); + _conn.getProtocolHandler().setFailoverLatch(null); + } } ExceptionListener listener = _conn._exceptionListener; |