This closes #409
This commit is contained in:
commit
d7d494e8f1
|
@ -516,6 +516,11 @@ public class ClientSessionFactoryImpl implements ClientSessionFactoryInternal, C
|
|||
// this is just a debug, since an interrupt is an expected event (in case of a shutdown)
|
||||
ActiveMQClientLogger.LOGGER.debug(e1.getMessage(), e1);
|
||||
}
|
||||
catch (Throwable t) {
|
||||
//for anything else just close so clients are un blocked
|
||||
close();
|
||||
throw t;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
|
@ -902,18 +907,19 @@ public class ClientSessionFactoryImpl implements ClientSessionFactoryInternal, C
|
|||
|
||||
//we check if we can actually connect.
|
||||
// we do it here as to receive the reply connection has to be not null
|
||||
//make sure to reset this.connection == null
|
||||
if (connection != null && liveNodeID != null) {
|
||||
try {
|
||||
if (!clientProtocolManager.checkForFailover(liveNodeID)) {
|
||||
connection.destroy();
|
||||
connection = null;
|
||||
this.connection = null;
|
||||
return null;
|
||||
}
|
||||
}
|
||||
catch (ActiveMQException e) {
|
||||
if (connection != null) {
|
||||
connection.destroy();
|
||||
connection = null;
|
||||
}
|
||||
connection.destroy();
|
||||
this.connection = null;
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
@ -665,6 +665,7 @@ public class NettyConnector extends AbstractConnector {
|
|||
ch.writeAndFlush(request);
|
||||
|
||||
if (!httpUpgradeHandler.awaitHandshake()) {
|
||||
ch.close().awaitUninterruptibly();
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
|
|
@ -76,6 +76,11 @@ public interface RemotingService {
|
|||
*/
|
||||
void pauseAcceptors();
|
||||
|
||||
/**
|
||||
* Pauses the acceptors so that no more connections can be made to the server
|
||||
*/
|
||||
boolean isPaused();
|
||||
|
||||
/**
|
||||
* Freezes and then disconnects all connections except the given one and tells the client where else
|
||||
* it might connect (only applicable if server is in a cluster and uses scaleDown-on-failover=true).
|
||||
|
|
|
@ -113,6 +113,8 @@ public class RemotingServiceImpl implements RemotingService, ServerConnectionLif
|
|||
|
||||
private ServiceRegistry serviceRegistry;
|
||||
|
||||
private boolean paused = false;
|
||||
|
||||
// Static --------------------------------------------------------
|
||||
|
||||
// Constructors --------------------------------------------------
|
||||
|
@ -192,6 +194,8 @@ public class RemotingServiceImpl implements RemotingService, ServerConnectionLif
|
|||
return;
|
||||
}
|
||||
|
||||
paused = false;
|
||||
|
||||
// The remoting service maintains it's own thread pool for handling remoting traffic
|
||||
// If OIO each connection will have it's own thread
|
||||
// If NIO these are capped at nio-remoting-threads which defaults to num cores * 3
|
||||
|
@ -316,6 +320,8 @@ public class RemotingServiceImpl implements RemotingService, ServerConnectionLif
|
|||
if (!started)
|
||||
return;
|
||||
|
||||
paused = true;
|
||||
|
||||
for (Acceptor acceptor : acceptors.values()) {
|
||||
try {
|
||||
acceptor.pause();
|
||||
|
@ -326,6 +332,11 @@ public class RemotingServiceImpl implements RemotingService, ServerConnectionLif
|
|||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public synchronized boolean isPaused() {
|
||||
return paused;
|
||||
}
|
||||
|
||||
@Override
|
||||
public synchronized void freeze(final String scaleDownNodeID, final CoreRemotingConnection connectionToKeepOpen) {
|
||||
if (!started)
|
||||
|
|
Loading…
Reference in New Issue