[hornetq-commits] JBoss hornetq SVN: r11371 - in branches/Branch_2_2_EAP: src/main/org/hornetq/core/protocol/core/impl and 15 other directories.

Author: clebert.suconic(a)jboss.com Date: 2011-09-19 20:50:32 -0400 (Mon, 19 Sep 2011) New Revision: 11371 Modified: branches/Branch_2_2_EAP/src/main/org/hornetq/core/client/impl/ClientSessionFactoryImpl.java branches/Branch_2_2_EAP/src/main/org/hornetq/core/client/impl/ServerLocatorImpl.java branches/Branch_2_2_EAP/src/main/org/hornetq/core/client/impl/ServerLocatorInternal.java branches/Branch_2_2_EAP/src/main/org/hornetq/core/client/impl/Topology.java branches/Branch_2_2_EAP/src/main/org/hornetq/core/protocol/core/impl/CoreProtocolManager.java branches/Branch_2_2_EAP/src/main/org/hornetq/core/protocol/stomp/StompProtocolManager.java branches/Branch_2_2_EAP/src/main/org/hornetq/core/remoting/impl/invm/InVMAcceptor.java branches/Branch_2_2_EAP/src/main/org/hornetq/core/remoting/impl/invm/InVMAcceptorFactory.java branches/Branch_2_2_EAP/src/main/org/hornetq/core/remoting/impl/invm/InVMConnection.java branches/Branch_2_2_EAP/src/main/org/hornetq/core/remoting/impl/invm/InVMConnector.java branches/Branch_2_2_EAP/src/main/org/hornetq/core/remoting/impl/netty/NettyAcceptor.java branches/Branch_2_2_EAP/src/main/org/hornetq/core/remoting/impl/netty/NettyAcceptorFactory.java branches/Branch_2_2_EAP/src/main/org/hornetq/core/remoting/impl/netty/NettyConnection.java branches/Branch_2_2_EAP/src/main/org/hornetq/core/remoting/impl/netty/NettyConnector.java branches/Branch_2_2_EAP/src/main/org/hornetq/core/remoting/server/impl/RemotingServiceImpl.java branches/Branch_2_2_EAP/src/main/org/hornetq/core/server/cluster/ClusterConnection.java branches/Branch_2_2_EAP/src/main/org/hornetq/core/server/cluster/ClusterManager.java branches/Branch_2_2_EAP/src/main/org/hornetq/core/server/cluster/impl/ClusterConnectionImpl.java branches/Branch_2_2_EAP/src/main/org/hornetq/core/server/cluster/impl/ClusterManagerImpl.java branches/Branch_2_2_EAP/src/main/org/hornetq/core/server/cluster/impl/ClusterManagerInternal.java branches/Branch_2_2_EAP/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java branches/Branch_2_2_EAP/src/main/org/hornetq/core/server/impl/ServerSessionImpl.java branches/Branch_2_2_EAP/src/main/org/hornetq/spi/core/protocol/ProtocolManager.java branches/Branch_2_2_EAP/src/main/org/hornetq/spi/core/remoting/Acceptor.java branches/Branch_2_2_EAP/src/main/org/hornetq/spi/core/remoting/AcceptorFactory.java branches/Branch_2_2_EAP/src/main/org/hornetq/spi/core/remoting/ConnectionLifeCycleListener.java branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/integration/cluster/distribution/ClusterTestBase.java branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/integration/cluster/distribution/OnewayTwoNodeClusterTest.java branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/integration/cluster/distribution/SimpleSymmetricClusterTest.java branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/integration/cluster/distribution/TwoWayTwoNodeClusterTest.java branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/integration/cluster/failover/FailBackAutoTest.java branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/integration/cluster/failover/GroupingFailoverTestBase.java branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/integration/cluster/failover/MultipleBackupsFailoverTestBase.java branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/integration/cluster/failover/MultipleLivesMultipleBackupsFailoverTest.java branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/integration/http/NettyHttpTest.java branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/integration/largemessage/mock/MockConnector.java branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/unit/core/remoting/impl/netty/NettyAcceptorFactoryTest.java branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/unit/core/remoting/impl/netty/NettyAcceptorTest.java branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/unit/core/remoting/impl/netty/NettyConnectionTest.java branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/unit/core/remoting/impl/netty/NettyConnectorTest.java branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/util/ServiceTestBase.java Log: Refactoring clustering manager / cluster connection Modified: branches/Branch_2_2_EAP/src/main/org/hornetq/core/client/impl/ClientSessionFactoryImpl.java =================================================================== --- branches/Branch_2_2_EAP/src/main/org/hornetq/core/client/impl/ClientSessionFactoryImpl.java 2011-09-19 23:47:54 UTC (rev 11370) +++ branches/Branch_2_2_EAP/src/main/org/hornetq/core/client/impl/ClientSessionFactoryImpl.java 2011-09-20 00:50:32 UTC (rev 11371) @@ -53,6 +53,7 @@ import org.hornetq.core.remoting.FailureListener; import org.hornetq.core.version.Version; import org.hornetq.spi.core.protocol.ProtocolType; +import org.hornetq.spi.core.remoting.Acceptor; import org.hornetq.spi.core.remoting.BufferHandler; import org.hornetq.spi.core.remoting.Connection; import org.hornetq.spi.core.remoting.ConnectionLifeCycleListener; @@ -367,7 +368,7 @@ // ConnectionLifeCycleListener implementation -------------------------------------------------- - public void connectionCreated(final Connection connection, final ProtocolType protocol) + public void connectionCreated(final Acceptor acceptor, final Connection connection, final ProtocolType protocol) { } Modified: branches/Branch_2_2_EAP/src/main/org/hornetq/core/client/impl/ServerLocatorImpl.java =================================================================== --- branches/Branch_2_2_EAP/src/main/org/hornetq/core/client/impl/ServerLocatorImpl.java 2011-09-19 23:47:54 UTC (rev 11370) +++ branches/Branch_2_2_EAP/src/main/org/hornetq/core/client/impl/ServerLocatorImpl.java 2011-09-20 00:50:32 UTC (rev 11371) @@ -1111,6 +1111,11 @@ } } + public String getIdentity() + { + return identity; + } + public void setIdentity(String identity) { this.identity = identity; @@ -1282,8 +1287,8 @@ { log.debug("nodeDown " + this + " nodeID=" + nodeID + " as being down", new Exception("trace")); } - - if (topology.removeMember(eventTime, nodeID)) + + if (!(isClusterConnection() && nodeID.equals(this.getNodeID())) && topology.removeMember(eventTime, nodeID)) { if (topology.isEmpty()) { Modified: branches/Branch_2_2_EAP/src/main/org/hornetq/core/client/impl/ServerLocatorInternal.java =================================================================== --- branches/Branch_2_2_EAP/src/main/org/hornetq/core/client/impl/ServerLocatorInternal.java 2011-09-19 23:47:54 UTC (rev 11370) +++ branches/Branch_2_2_EAP/src/main/org/hornetq/core/client/impl/ServerLocatorInternal.java 2011-09-20 00:50:32 UTC (rev 11371) @@ -39,8 +39,12 @@ void setAfterConnectionInternalListener(AfterConnectInternalListener listener); - /** Used to better identify Cluster Connection Locators on logs while debugging logs */ + /** Used to better identify Cluster Connection Locators on logs. To facilitate eventual debugging. + * + * This method used to be on tests interface, but I'm now making it part of the public interface since*/ void setIdentity(String identity); + + String getIdentity(); void setNodeID(String nodeID); Modified: branches/Branch_2_2_EAP/src/main/org/hornetq/core/client/impl/Topology.java =================================================================== --- branches/Branch_2_2_EAP/src/main/org/hornetq/core/client/impl/Topology.java 2011-09-19 23:47:54 UTC (rev 11370) +++ branches/Branch_2_2_EAP/src/main/org/hornetq/core/client/impl/Topology.java 2011-09-20 00:50:32 UTC (rev 11371) @@ -108,7 +108,7 @@ { if (log.isDebugEnabled()) { - log.info(this + "::Live node " + nodeId + "=" + memberInput); + log.debug(this + "::node " + nodeId + "=" + memberInput); } memberInput.setUniqueEventID(System.currentTimeMillis()); mapTopology.remove(nodeId); @@ -212,7 +212,7 @@ currentMember + ", memberInput=" + memberInput + - "newMember=" + newMember); + "newMember=" + newMember, new Exception ("trace")); } @@ -301,7 +301,7 @@ { if (member.getUniqueEventID() > uniqueEventID) { - log.info("The removeMember was issued before the node " + nodeId + " was started, ignoring call"); + log.debug("The removeMember was issued before the node " + nodeId + " was started, ignoring call"); member = null; } else @@ -482,22 +482,17 @@ public synchronized String describe(final String text) { - String desc = text + "\n"; + String desc = text + "topology on " + this + ":\n"; for (Entry<String, TopologyMember> entry : new HashMap<String, TopologyMember>(mapTopology).entrySet()) { desc += "\t" + entry.getKey() + " => " + entry.getValue() + "\n"; } desc += "\t" + "nodes=" + nodes() + "\t" + "members=" + members(); - return desc; - } - - public void clear() - { - if (Topology.log.isDebugEnabled()) + if (mapTopology.isEmpty()) { - Topology.log.debug(this + "::clear", new Exception("trace")); + desc += "\tEmpty"; } - mapTopology.clear(); + return desc; } public int members() Modified: branches/Branch_2_2_EAP/src/main/org/hornetq/core/protocol/core/impl/CoreProtocolManager.java =================================================================== --- branches/Branch_2_2_EAP/src/main/org/hornetq/core/protocol/core/impl/CoreProtocolManager.java 2011-09-19 23:47:54 UTC (rev 11370) +++ branches/Branch_2_2_EAP/src/main/org/hornetq/core/protocol/core/impl/CoreProtocolManager.java 2011-09-20 00:50:32 UTC (rev 11371) @@ -42,6 +42,7 @@ import org.hornetq.spi.core.protocol.ConnectionEntry; import org.hornetq.spi.core.protocol.ProtocolManager; import org.hornetq.spi.core.protocol.RemotingConnection; +import org.hornetq.spi.core.remoting.Acceptor; import org.hornetq.spi.core.remoting.Connection; /** @@ -68,7 +69,7 @@ this.interceptors = interceptors; } - public ConnectionEntry createConnectionEntry(final Connection connection) + public ConnectionEntry createConnectionEntry(final Acceptor acceptorUsed, final Connection connection) { final Configuration config = server.getConfiguration(); @@ -177,16 +178,18 @@ }; final boolean isCC = msg.isClusterConnection(); - - server.getClusterManager().addClusterTopologyListener(listener, isCC); - - rc.addCloseListener(new CloseListener() + if (acceptorUsed.getClusterConnection() != null) { - public void connectionClosed() + acceptorUsed.getClusterConnection().addClusterTopologyListener(listener, isCC); + + rc.addCloseListener(new CloseListener() { - server.getClusterManager().removeClusterTopologyListener(listener, isCC); - } - }); + public void connectionClosed() + { + acceptorUsed.getClusterConnection().removeClusterTopologyListener(listener, isCC); + } + }); + } } else if (packet.getType() == PacketImpl.NODE_ANNOUNCE) { @@ -205,7 +208,8 @@ { log.trace("Server " + server + " receiving nodeUp from NodeID=" + msg.getNodeID() + ", pair=" + pair); } - server.getClusterManager().nodeAnnounced(msg.getCurrentEventID(), msg.getNodeID(), pair, msg.isBackup()); + + acceptorUsed.getClusterConnection().nodeAnnounced(msg.getCurrentEventID(), msg.getNodeID(), pair, msg.isBackup()); } } }); Modified: branches/Branch_2_2_EAP/src/main/org/hornetq/core/protocol/stomp/StompProtocolManager.java =================================================================== --- branches/Branch_2_2_EAP/src/main/org/hornetq/core/protocol/stomp/StompProtocolManager.java 2011-09-19 23:47:54 UTC (rev 11370) +++ branches/Branch_2_2_EAP/src/main/org/hornetq/core/protocol/stomp/StompProtocolManager.java 2011-09-20 00:50:32 UTC (rev 11371) @@ -38,6 +38,7 @@ import org.hornetq.spi.core.protocol.ConnectionEntry; import org.hornetq.spi.core.protocol.ProtocolManager; import org.hornetq.spi.core.protocol.RemotingConnection; +import org.hornetq.spi.core.remoting.Acceptor; import org.hornetq.spi.core.remoting.Connection; import org.hornetq.spi.core.security.HornetQSecurityManager; import org.hornetq.utils.UUIDGenerator; @@ -109,7 +110,7 @@ // ProtocolManager implementation -------------------------------- - public ConnectionEntry createConnectionEntry(final Connection connection) + public ConnectionEntry createConnectionEntry(final Acceptor acceptorUsed, final Connection connection) { StompConnection conn = new StompConnection(connection, this); Modified: branches/Branch_2_2_EAP/src/main/org/hornetq/core/remoting/impl/invm/InVMAcceptor.java =================================================================== --- branches/Branch_2_2_EAP/src/main/org/hornetq/core/remoting/impl/invm/InVMAcceptor.java 2011-09-19 23:47:54 UTC (rev 11370) +++ branches/Branch_2_2_EAP/src/main/org/hornetq/core/remoting/impl/invm/InVMAcceptor.java 2011-09-20 00:50:32 UTC (rev 11371) @@ -21,6 +21,7 @@ import org.hornetq.api.core.SimpleString; import org.hornetq.api.core.management.NotificationType; import org.hornetq.core.logging.Logger; +import org.hornetq.core.server.cluster.ClusterConnection; import org.hornetq.core.server.management.Notification; import org.hornetq.core.server.management.NotificationService; import org.hornetq.spi.core.protocol.ProtocolType; @@ -54,16 +55,21 @@ private volatile boolean started; private final ExecutorFactory executorFactory; + + private final ClusterConnection clusterConnection; private boolean paused; private NotificationService notificationService; - public InVMAcceptor(final Map<String, Object> configuration, + public InVMAcceptor(final ClusterConnection clusterConnection, + final Map<String, Object> configuration, final BufferHandler handler, final ConnectionLifeCycleListener listener, final Executor threadPool) { + this.clusterConnection = clusterConnection; + this.handler = handler; this.listener = listener; @@ -73,6 +79,11 @@ executorFactory = new OrderedExecutorFactory(threadPool); } + public ClusterConnection getClusterConnection() + { + return clusterConnection; + } + public synchronized void start() throws Exception { if (started) @@ -189,7 +200,7 @@ throw new IllegalStateException("Acceptor is not started"); } - new InVMConnection(id, connectionID, remoteHandler, new Listener(connector), clientExecutor); + new InVMConnection(this, id, connectionID, remoteHandler, new Listener(connector), clientExecutor); } public void disconnect(final String connectionID) @@ -209,6 +220,8 @@ private class Listener implements ConnectionLifeCycleListener { + //private static Listener instance = new Listener(); + private final InVMConnector connector; Listener(final InVMConnector connector) @@ -216,14 +229,14 @@ this.connector = connector; } - public void connectionCreated(final Connection connection, final ProtocolType protocol) + public void connectionCreated(final Acceptor acceptor, final Connection connection, final ProtocolType protocol) { if (connections.putIfAbsent((String)connection.getID(), connection) != null) { throw new IllegalArgumentException("Connection already exists with id " + connection.getID()); } - listener.connectionCreated(connection, protocol); + listener.connectionCreated(acceptor, connection, protocol); } public void connectionDestroyed(final Object connectionID) Modified: branches/Branch_2_2_EAP/src/main/org/hornetq/core/remoting/impl/invm/InVMAcceptorFactory.java =================================================================== --- branches/Branch_2_2_EAP/src/main/org/hornetq/core/remoting/impl/invm/InVMAcceptorFactory.java 2011-09-19 23:47:54 UTC (rev 11370) +++ branches/Branch_2_2_EAP/src/main/org/hornetq/core/remoting/impl/invm/InVMAcceptorFactory.java 2011-09-20 00:50:32 UTC (rev 11371) @@ -17,6 +17,7 @@ import java.util.concurrent.Executor; import java.util.concurrent.ScheduledExecutorService; +import org.hornetq.core.server.cluster.ClusterConnection; import org.hornetq.spi.core.remoting.Acceptor; import org.hornetq.spi.core.remoting.AcceptorFactory; import org.hornetq.spi.core.remoting.BufferDecoder; @@ -31,14 +32,15 @@ */ public class InVMAcceptorFactory implements AcceptorFactory { - public Acceptor createAcceptor(final Map<String, Object> configuration, + public Acceptor createAcceptor(final ClusterConnection clusterConnection, + final Map<String, Object> configuration, final BufferHandler handler, final BufferDecoder decoder, final ConnectionLifeCycleListener listener, final Executor threadPool, final ScheduledExecutorService scheduledThreadPool) { - return new InVMAcceptor(configuration, handler, listener, threadPool); + return new InVMAcceptor(clusterConnection, configuration, handler, listener, threadPool); } public Set<String> getAllowableProperties() Modified: branches/Branch_2_2_EAP/src/main/org/hornetq/core/remoting/impl/invm/InVMConnection.java =================================================================== --- branches/Branch_2_2_EAP/src/main/org/hornetq/core/remoting/impl/invm/InVMConnection.java 2011-09-19 23:47:54 UTC (rev 11370) +++ branches/Branch_2_2_EAP/src/main/org/hornetq/core/remoting/impl/invm/InVMConnection.java 2011-09-20 00:50:32 UTC (rev 11371) @@ -21,6 +21,7 @@ import org.hornetq.api.core.HornetQBuffers; import org.hornetq.core.logging.Logger; import org.hornetq.spi.core.protocol.ProtocolType; +import org.hornetq.spi.core.remoting.Acceptor; import org.hornetq.spi.core.remoting.BufferHandler; import org.hornetq.spi.core.remoting.Connection; import org.hornetq.spi.core.remoting.ConnectionLifeCycleListener; @@ -57,15 +58,17 @@ private volatile boolean closing; - public InVMConnection(final int serverID, + public InVMConnection(final Acceptor acceptor, + final int serverID, final BufferHandler handler, final ConnectionLifeCycleListener listener, final Executor executor) { - this(serverID, UUIDGenerator.getInstance().generateSimpleStringUUID().toString(), handler, listener, executor); + this(acceptor, serverID, UUIDGenerator.getInstance().generateSimpleStringUUID().toString(), handler, listener, executor); } - public InVMConnection(final int serverID, + public InVMConnection(final Acceptor acceptor, + final int serverID, final String id, final BufferHandler handler, final ConnectionLifeCycleListener listener, @@ -81,7 +84,7 @@ this.executor = executor; - listener.connectionCreated(this, ProtocolType.CORE); + listener.connectionCreated(acceptor, this, ProtocolType.CORE); } public void close() Modified: branches/Branch_2_2_EAP/src/main/org/hornetq/core/remoting/impl/invm/InVMConnector.java =================================================================== --- branches/Branch_2_2_EAP/src/main/org/hornetq/core/remoting/impl/invm/InVMConnector.java 2011-09-19 23:47:54 UTC (rev 11370) +++ branches/Branch_2_2_EAP/src/main/org/hornetq/core/remoting/impl/invm/InVMConnector.java 2011-09-20 00:50:32 UTC (rev 11371) @@ -172,19 +172,20 @@ final ConnectionLifeCycleListener listener, final Executor serverExecutor) { - return new InVMConnection(id, handler, listener, serverExecutor); + // No acceptor on a client connection + return new InVMConnection(null, id, handler, listener, serverExecutor); } private class Listener implements ConnectionLifeCycleListener { - public void connectionCreated(final Connection connection, final ProtocolType protocol) + public void connectionCreated(final Acceptor acceptor, final Connection connection, final ProtocolType protocol) { if (connections.putIfAbsent((String)connection.getID(), connection) != null) { throw new IllegalArgumentException("Connection already exists with id " + connection.getID()); } - listener.connectionCreated(connection, protocol); + listener.connectionCreated(acceptor, connection, protocol); } public void connectionDestroyed(final Object connectionID) Modified: branches/Branch_2_2_EAP/src/main/org/hornetq/core/remoting/impl/netty/NettyAcceptor.java =================================================================== --- branches/Branch_2_2_EAP/src/main/org/hornetq/core/remoting/impl/netty/NettyAcceptor.java 2011-09-19 23:47:54 UTC (rev 11370) +++ branches/Branch_2_2_EAP/src/main/org/hornetq/core/remoting/impl/netty/NettyAcceptor.java 2011-09-20 00:50:32 UTC (rev 11371) @@ -37,6 +37,7 @@ import org.hornetq.core.logging.Logger; import org.hornetq.core.protocol.stomp.WebSocketServerHandler; import org.hornetq.core.remoting.impl.ssl.SSLSupport; +import org.hornetq.core.server.cluster.ClusterConnection; import org.hornetq.core.server.management.Notification; import org.hornetq.core.server.management.NotificationService; import org.hornetq.spi.core.protocol.ProtocolType; @@ -87,6 +88,8 @@ { static final Logger log = Logger.getLogger(NettyAcceptor.class); + private ClusterConnection clusterConnection; + private ChannelFactory channelFactory; private volatile ChannelGroup serverChannelGroup; @@ -158,6 +161,7 @@ private final long batchDelay; private final boolean directDeliver; + public NettyAcceptor(final Map<String, Object> configuration, final BufferHandler handler, @@ -166,6 +170,21 @@ final Executor threadPool, final ScheduledExecutorService scheduledThreadPool) { + this(null, configuration, handler, decoder, listener, threadPool, scheduledThreadPool); + } + + + public NettyAcceptor(final ClusterConnection clusterConnection, + final Map<String, Object> configuration, + final BufferHandler handler, + final BufferDecoder decoder, + final ConnectionLifeCycleListener listener, + final Executor threadPool, + final ScheduledExecutorService scheduledThreadPool) + { + + this.clusterConnection = clusterConnection; + this.handler = handler; this.decoder = decoder; @@ -618,6 +637,14 @@ { this.notificationService = notificationService; } + + /* (non-Javadoc) + * @see org.hornetq.spi.core.remoting.Acceptor#getClusterConnection() + */ + public ClusterConnection getClusterConnection() + { + return clusterConnection; + } // Inner classes ----------------------------------------------------------------------------- @@ -633,7 +660,7 @@ @Override public void channelConnected(final ChannelHandlerContext ctx, final ChannelStateEvent e) throws Exception { - new NettyConnection(e.getChannel(), new Listener(), !httpEnabled && batchDelay > 0, directDeliver); + new NettyConnection(NettyAcceptor.this, e.getChannel(), new Listener(), !httpEnabled && batchDelay > 0, directDeliver); SslHandler sslHandler = ctx.getPipeline().get(SslHandler.class); if (sslHandler != null) @@ -662,14 +689,14 @@ private class Listener implements ConnectionLifeCycleListener { - public void connectionCreated(final Connection connection, final ProtocolType protocol) + public void connectionCreated(final Acceptor acceptor, final Connection connection, final ProtocolType protocol) { if (connections.putIfAbsent(connection.getID(), (NettyConnection)connection) != null) { throw new IllegalArgumentException("Connection already exists with id " + connection.getID()); } - listener.connectionCreated(connection, NettyAcceptor.this.protocol); + listener.connectionCreated(acceptor, connection, NettyAcceptor.this.protocol); } public void connectionDestroyed(final Object connectionID) Modified: branches/Branch_2_2_EAP/src/main/org/hornetq/core/remoting/impl/netty/NettyAcceptorFactory.java =================================================================== --- branches/Branch_2_2_EAP/src/main/org/hornetq/core/remoting/impl/netty/NettyAcceptorFactory.java 2011-09-19 23:47:54 UTC (rev 11370) +++ branches/Branch_2_2_EAP/src/main/org/hornetq/core/remoting/impl/netty/NettyAcceptorFactory.java 2011-09-20 00:50:32 UTC (rev 11371) @@ -18,6 +18,7 @@ import java.util.concurrent.Executor; import java.util.concurrent.ScheduledExecutorService; +import org.hornetq.core.server.cluster.ClusterConnection; import org.hornetq.spi.core.remoting.Acceptor; import org.hornetq.spi.core.remoting.AcceptorFactory; import org.hornetq.spi.core.remoting.BufferDecoder; @@ -31,14 +32,15 @@ */ public class NettyAcceptorFactory implements AcceptorFactory { - public Acceptor createAcceptor(final Map<String, Object> configuration, + public Acceptor createAcceptor(final ClusterConnection connection, + final Map<String, Object> configuration, final BufferHandler handler, final BufferDecoder decoder, final ConnectionLifeCycleListener listener, final Executor threadPool, final ScheduledExecutorService scheduledThreadPool) { - return new NettyAcceptor(configuration, handler, decoder, listener, threadPool, scheduledThreadPool); + return new NettyAcceptor(connection, configuration, handler, decoder, listener, threadPool, scheduledThreadPool); } public Set<String> getAllowableProperties() Modified: branches/Branch_2_2_EAP/src/main/org/hornetq/core/remoting/impl/netty/NettyConnection.java =================================================================== --- branches/Branch_2_2_EAP/src/main/org/hornetq/core/remoting/impl/netty/NettyConnection.java 2011-09-19 23:47:54 UTC (rev 11370) +++ branches/Branch_2_2_EAP/src/main/org/hornetq/core/remoting/impl/netty/NettyConnection.java 2011-09-20 00:50:32 UTC (rev 11371) @@ -21,6 +21,7 @@ import org.hornetq.core.buffers.impl.ChannelBufferWrapper; import org.hornetq.core.logging.Logger; import org.hornetq.spi.core.protocol.ProtocolType; +import org.hornetq.spi.core.remoting.Acceptor; import org.hornetq.spi.core.remoting.Connection; import org.hornetq.spi.core.remoting.ConnectionLifeCycleListener; import org.hornetq.spi.core.remoting.ReadyListener; @@ -72,6 +73,15 @@ boolean batchingEnabled, boolean directDeliver) { + this(null, channel, listener, batchingEnabled, directDeliver); + } + + public NettyConnection(final Acceptor acceptor, + final Channel channel, + final ConnectionLifeCycleListener listener, + boolean batchingEnabled, + boolean directDeliver) + { this.channel = channel; this.listener = listener; @@ -80,7 +90,7 @@ this.directDeliver = directDeliver; - listener.connectionCreated(this, ProtocolType.CORE); + listener.connectionCreated(acceptor, this, ProtocolType.CORE); } // Public -------------------------------------------------------- Modified: branches/Branch_2_2_EAP/src/main/org/hornetq/core/remoting/impl/netty/NettyConnector.java =================================================================== --- branches/Branch_2_2_EAP/src/main/org/hornetq/core/remoting/impl/netty/NettyConnector.java 2011-09-19 23:47:54 UTC (rev 11370) +++ branches/Branch_2_2_EAP/src/main/org/hornetq/core/remoting/impl/netty/NettyConnector.java 2011-09-20 00:50:32 UTC (rev 11371) @@ -35,6 +35,7 @@ import org.hornetq.core.logging.Logger; import org.hornetq.core.remoting.impl.ssl.SSLSupport; import org.hornetq.spi.core.protocol.ProtocolType; +import org.hornetq.spi.core.remoting.Acceptor; import org.hornetq.spi.core.remoting.BufferHandler; import org.hornetq.spi.core.remoting.Connection; import org.hornetq.spi.core.remoting.ConnectionLifeCycleListener; @@ -495,7 +496,8 @@ ch.getPipeline().get(HornetQChannelHandler.class).active = true; } - NettyConnection conn = new NettyConnection(ch, new Listener(), !httpEnabled && batchDelay > 0, false); + // No acceptor on a client connection + NettyConnection conn = new NettyConnection(null, ch, new Listener(), !httpEnabled && batchDelay > 0, false); return conn; } @@ -689,7 +691,7 @@ private class Listener implements ConnectionLifeCycleListener { - public void connectionCreated(final Connection connection, final ProtocolType protocol) + public void connectionCreated(final Acceptor acceptor, final Connection connection, final ProtocolType protocol) { if (connections.putIfAbsent(connection.getID(), connection) != null) { Modified: branches/Branch_2_2_EAP/src/main/org/hornetq/core/remoting/server/impl/RemotingServiceImpl.java =================================================================== --- branches/Branch_2_2_EAP/src/main/org/hornetq/core/remoting/server/impl/RemotingServiceImpl.java 2011-09-19 23:47:54 UTC (rev 11370) +++ branches/Branch_2_2_EAP/src/main/org/hornetq/core/remoting/server/impl/RemotingServiceImpl.java 2011-09-20 00:50:32 UTC (rev 11371) @@ -39,6 +39,7 @@ import org.hornetq.core.remoting.impl.netty.TransportConstants; import org.hornetq.core.remoting.server.RemotingService; import org.hornetq.core.server.HornetQServer; +import org.hornetq.core.server.cluster.ClusterManager; import org.hornetq.core.server.impl.ServerSessionImpl; import org.hornetq.core.server.management.ManagementService; import org.hornetq.spi.core.protocol.ConnectionEntry; @@ -94,6 +95,8 @@ private final ScheduledExecutorService scheduledThreadPool; private FailureCheckAndFlushThread failureCheckAndFlushThread; + + private final ClusterManager clusterManager; private Map<ProtocolType, ProtocolManager> protocolMap = new ConcurrentHashMap<ProtocolType, ProtocolManager>(); @@ -101,7 +104,8 @@ // Constructors -------------------------------------------------- - public RemotingServiceImpl(final Configuration config, + public RemotingServiceImpl(final ClusterManager clusterManager, + final Configuration config, final HornetQServer server, final ManagementService managementService, final ScheduledExecutorService scheduledThreadPool) @@ -109,6 +113,8 @@ transportConfigs = config.getAcceptorConfigurations(); this.server = server; + + this.clusterManager = clusterManager; ClassLoader loader = Thread.currentThread().getContextClassLoader(); for (String interceptorClass : config.getInterceptorClassNames()) @@ -202,7 +208,9 @@ ProtocolManager manager = protocolMap.get(protocol); - Acceptor acceptor = factory.createAcceptor(info.getParams(), + // TODO: parameterize the cluster connection + Acceptor acceptor = factory.createAcceptor(clusterManager.getDefaultConnection(), + info.getParams(), new DelegatingBufferHandler(), manager, this, @@ -370,7 +378,7 @@ return protocolMap.get(protocol); } - public void connectionCreated(final Connection connection, final ProtocolType protocol) + public void connectionCreated(final Acceptor acceptor, final Connection connection, final ProtocolType protocol) { if (server == null) { @@ -384,7 +392,7 @@ throw new IllegalArgumentException("Unknown protocol " + protocol); } - ConnectionEntry entry = pmgr.createConnectionEntry(connection); + ConnectionEntry entry = pmgr.createConnectionEntry(acceptor, connection); if (isTrace) { Modified: branches/Branch_2_2_EAP/src/main/org/hornetq/core/server/cluster/ClusterConnection.java =================================================================== --- branches/Branch_2_2_EAP/src/main/org/hornetq/core/server/cluster/ClusterConnection.java 2011-09-19 23:47:54 UTC (rev 11370) +++ branches/Branch_2_2_EAP/src/main/org/hornetq/core/server/cluster/ClusterConnection.java 2011-09-20 00:50:32 UTC (rev 11371) @@ -15,9 +15,11 @@ import java.util.Map; +import org.hornetq.api.core.Pair; import org.hornetq.api.core.SimpleString; import org.hornetq.api.core.TransportConfiguration; import org.hornetq.api.core.client.ClusterTopologyListener; +import org.hornetq.core.client.impl.Topology; import org.hornetq.core.server.HornetQComponent; import org.hornetq.core.server.HornetQServer; @@ -37,7 +39,13 @@ String getNodeID(); HornetQServer getServer(); + + void nodeAnnounced(long eventUID, String nodeID, Pair<TransportConfiguration, TransportConfiguration> connectorPair, boolean backup); + void addClusterTopologyListener(ClusterTopologyListener listener, boolean clusterConnection); + + void removeClusterTopologyListener(ClusterTopologyListener listener, boolean clusterConnection); + /** * @return a Map of node ID and addresses */ @@ -47,8 +55,14 @@ TransportConfiguration getConnector(); + Topology getTopology(); + void flushExecutor(); // for debug String describe(); + + void informTopology(); + + void announceBackup(); } Modified: branches/Branch_2_2_EAP/src/main/org/hornetq/core/server/cluster/ClusterManager.java =================================================================== --- branches/Branch_2_2_EAP/src/main/org/hornetq/core/server/cluster/ClusterManager.java 2011-09-19 23:47:54 UTC (rev 11370) +++ branches/Branch_2_2_EAP/src/main/org/hornetq/core/server/cluster/ClusterManager.java 2011-09-20 00:50:32 UTC (rev 11371) @@ -16,11 +16,7 @@ import java.util.Map; import java.util.Set; -import org.hornetq.api.core.Pair; import org.hornetq.api.core.SimpleString; -import org.hornetq.api.core.TransportConfiguration; -import org.hornetq.api.core.client.ClusterTopologyListener; -import org.hornetq.core.client.impl.Topology; import org.hornetq.core.config.BridgeConfiguration; import org.hornetq.core.server.HornetQComponent; @@ -37,26 +33,26 @@ Map<String, Bridge> getBridges(); Set<ClusterConnection> getClusterConnections(); + + /** + * Return the default ClusterConnection to be used case it's not defined by the acceptor + * @return + */ + ClusterConnection getDefaultConnection(); ClusterConnection getClusterConnection(SimpleString name); Set<BroadcastGroup> getBroadcastGroups(); - - void addClusterTopologyListener(ClusterTopologyListener listener, boolean clusterConnection); - void removeClusterTopologyListener(ClusterTopologyListener listener, boolean clusterConnection); - void activate(); - void nodeAnnounced(long eventUID, String nodeID, Pair<TransportConfiguration, TransportConfiguration> connectorPair, boolean backup); - - Topology getTopology(); - void flushExecutor(); void announceBackup() throws Exception; + + void deploy() throws Exception; - void deployBridge(BridgeConfiguration config) throws Exception; + void deployBridge(BridgeConfiguration config, boolean start) throws Exception; void destroyBridge(String name) throws Exception; Modified: branches/Branch_2_2_EAP/src/main/org/hornetq/core/server/cluster/impl/ClusterConnectionImpl.java =================================================================== --- branches/Branch_2_2_EAP/src/main/org/hornetq/core/server/cluster/impl/ClusterConnectionImpl.java 2011-09-19 23:47:54 UTC (rev 11370) +++ branches/Branch_2_2_EAP/src/main/org/hornetq/core/server/cluster/impl/ClusterConnectionImpl.java 2011-09-20 00:50:32 UTC (rev 11371) @@ -33,6 +33,8 @@ import org.hornetq.api.core.SimpleString; import org.hornetq.api.core.TransportConfiguration; import org.hornetq.api.core.client.ClientMessage; +import org.hornetq.api.core.client.ClientSessionFactory; +import org.hornetq.api.core.client.ClusterTopologyListener; import org.hornetq.api.core.management.ManagementHelper; import org.hornetq.api.core.management.NotificationType; import org.hornetq.core.client.impl.AfterConnectInternalListener; @@ -46,6 +48,7 @@ import org.hornetq.core.postoffice.Bindings; import org.hornetq.core.postoffice.PostOffice; import org.hornetq.core.postoffice.impl.PostOfficeImpl; +import org.hornetq.core.protocol.core.impl.wireformat.NodeAnnounceMessage; import org.hornetq.core.server.HornetQServer; import org.hornetq.core.server.Queue; import org.hornetq.core.server.cluster.Bridge; @@ -79,8 +82,6 @@ private final ExecutorFactory executorFactory; - private final Topology clusterManagerTopology; - private final Executor executor; private final HornetQServer server; @@ -140,9 +141,17 @@ private final Set<TransportConfiguration> allowableConnections = new HashSet<TransportConfiguration>(); private final ClusterManagerInternal manager; + + + // Stuff that used to be on the ClusterManager + + private final Topology topology = new Topology(this); + + private volatile ServerLocatorInternal backupServerLocator; + + public ClusterConnectionImpl(final ClusterManagerInternal manager, - final Topology clusterManagerTopology, final TransportConfiguration[] tcConfigs, final TransportConfiguration connector, final SimpleString name, @@ -204,6 +213,8 @@ this.executorFactory = executorFactory; this.executor = executorFactory.getExecutor(); + + this.topology.setExecutor(executor); this.server = server; @@ -227,8 +238,6 @@ this.callTimeout = callTimeout; - this.clusterManagerTopology = clusterManagerTopology; - clusterConnector = new StaticClusterConnector(tcConfigs); if (tcConfigs != null && tcConfigs.length > 0) @@ -244,7 +253,6 @@ } public ClusterConnectionImpl(final ClusterManagerImpl manager, - final Topology clusterManagerTopology, DiscoveryGroupConfiguration dg, final TransportConfiguration connector, final SimpleString name, @@ -308,6 +316,8 @@ this.executorFactory = executorFactory; this.executor = executorFactory.getExecutor(); + + this.topology.setExecutor(executor); this.server = server; @@ -330,11 +340,9 @@ clusterConnector = new DiscoveryClusterConnector(dg); this.manager = manager; - - this.clusterManagerTopology = clusterManagerTopology; } - public void start() throws Exception + public void start() throws Exception { synchronized (this) { @@ -410,13 +418,21 @@ props); managementService.sendNotification(notification); } + + executor.execute(new Runnable() { public void run() { synchronized (ClusterConnectionImpl.this) { + if (backupServerLocator != null) + { + backupServerLocator.close(); + backupServerLocator = null; + } + if (serverLocator != null) { serverLocator.close(); @@ -430,12 +446,97 @@ started = false; } + + public void announceBackup() + { + this.backupServerLocator = clusterConnector.createServerLocator(false); + + backupServerLocator.setReconnectAttempts(-1); + backupServerLocator.setInitialConnectAttempts(-1); + + + executor.execute(new Runnable() + { + public void run() + { + try + { + if (log.isDebugEnabled()) + { + log.debug(ClusterConnectionImpl.this + ":: announcing " + connector + " to " + backupServerLocator); + } + ClientSessionFactory backupSessionFactory = backupServerLocator.connect(); + if (backupSessionFactory != null) + { + backupSessionFactory.getConnection() + .getChannel(0, -1) + .send(new NodeAnnounceMessage(System.currentTimeMillis(), + nodeUUID.toString(), + true, + connector, + null)); + log.info("backup announced"); + } + } + catch (Exception e) + { + log.warn("Unable to announce backup, retrying", e); + } + } + }); + } + + private TopologyMember getLocalMember() + { + return topology.getMember(manager.getNodeId()); + } + + public void addClusterTopologyListener(final ClusterTopologyListener listener, final boolean clusterConnection) + { + topology.addClusterTopologyListener(listener); + + // no need to use an executor here since the Topology is already using one + topology.sendTopology(listener); + } + + public void removeClusterTopologyListener(final ClusterTopologyListener listener, final boolean clusterConnection) + { + topology.removeClusterTopologyListener(listener); + } + + public Topology getTopology() + { + return topology; + } + + public void nodeAnnounced(final long uniqueEventID, + final String nodeID, + final Pair<TransportConfiguration, TransportConfiguration> connectorPair, + final boolean backup) + { + if (log.isDebugEnabled()) + { + log.debug(this + "::NodeAnnounced, backup=" + backup + nodeID + connectorPair); + } + + TopologyMember newMember = new TopologyMember(connectorPair.a, connectorPair.b); + newMember.setUniqueEventID(uniqueEventID); + if (backup) + { + topology.updateBackup(nodeID, new TopologyMember(connectorPair.a, connectorPair.b)); + } + else + { + topology.updateMember(uniqueEventID, nodeID, newMember); + } + } + /* (non-Javadoc) * @see org.hornetq.core.client.impl.AfterConnectInternalListener#onConnection(org.hornetq.core.client.impl.ClientSessionFactoryInternal) */ public void onConnection(ClientSessionFactoryInternal sf) { - TopologyMember localMember = manager.getLocalMember(); + TopologyMember localMember = getLocalMember(); sf.sendNodeAnnounce(localMember.getUniqueEventID(), manager.getNodeId(), false, @@ -498,9 +599,27 @@ } backup = false; + + topology.updateAsLive(manager.getNodeId(), new TopologyMember(connector, null)); - serverLocator = clusterConnector.createServerLocator(); + if (backupServerLocator != null) + { + // todo we could use the topology of this to preempt it arriving from the cc + try + { + backupServerLocator.close(); + } + catch (Exception e) + { + log.warn("problem closing backup session factory", e); + } + backupServerLocator = null; + } + + + serverLocator = clusterConnector.createServerLocator(true); + if (serverLocator != null) { @@ -509,7 +628,7 @@ log.debug("DuplicateDetection is disabled, sending clustered messages blocked"); } - final TopologyMember currentMember = clusterManagerTopology.getMember(nodeUUID.toString()); + final TopologyMember currentMember = topology.getMember(manager.getNodeId()); if (currentMember == null) { @@ -554,6 +673,7 @@ log.debug("sending notification: " + notification); managementService.sendNotification(notification); } + } public TransportConfiguration getConnector() @@ -660,7 +780,6 @@ { log.debug(this + "::Creating record for nodeID=" + nodeID + ", connectorPair=" + connectorPair); } - log.info(this + "::Creating record for nodeID=" + nodeID + ", connectorPair=" + connectorPair); // New node - create a new flow record @@ -701,7 +820,26 @@ } } } + + public synchronized void informTopology() + { + String nodeID = server.getNodeID().toString(); + + TopologyMember localMember; + + if (backup) + { + localMember = new TopologyMember(null, connector); + } + else + { + localMember = new TopologyMember(connector, null); + } + topology.updateAsLive(nodeID, localMember); + } + + private void createNewRecord(final long eventUID, final String targetNodeID, final TransportConfiguration connector, @@ -709,7 +847,7 @@ final Queue queue, final boolean start) throws Exception { - final ServerLocatorInternal targetLocator = new ServerLocatorImpl(clusterManagerTopology, false, connector); + final ServerLocatorInternal targetLocator = new ServerLocatorImpl(topology, false, connector); String nodeId; @@ -1365,7 +1503,8 @@ @Override public String toString() { - return "ClusterConnectionImpl [nodeUUID=" + nodeUUID + + return "ClusterConnectionImpl@" + System.identityHashCode(this) + + "[nodeUUID=" + nodeUUID + ", connector=" + connector + ", address=" + @@ -1395,7 +1534,7 @@ interface ClusterConnector { - ServerLocatorInternal createServerLocator(); + ServerLocatorInternal createServerLocator(boolean includeTopology); } private class StaticClusterConnector implements ClusterConnector @@ -1407,7 +1546,7 @@ this.tcConfigs = tcConfigs; } - public ServerLocatorInternal createServerLocator() + public ServerLocatorInternal createServerLocator(boolean includeTopology) { if (tcConfigs != null && tcConfigs.length > 0) { @@ -1415,7 +1554,9 @@ { log.debug(ClusterConnectionImpl.this + "Creating a serverLocator for " + Arrays.toString(tcConfigs)); } - return new ServerLocatorImpl(clusterManagerTopology, true, tcConfigs); + ServerLocatorImpl locator = new ServerLocatorImpl(includeTopology?topology:null, true, tcConfigs); + locator.setClusterConnection(true); + return locator; } else { @@ -1443,9 +1584,11 @@ this.dg = dg; } - public ServerLocatorInternal createServerLocator() + public ServerLocatorInternal createServerLocator(boolean includeTopology) { - return new ServerLocatorImpl(clusterManagerTopology, true, dg); + ServerLocatorImpl locator = new ServerLocatorImpl(includeTopology?topology:null, true, dg); + return locator; + } } } Modified: branches/Branch_2_2_EAP/src/main/org/hornetq/core/server/cluster/impl/ClusterManagerImpl.java =================================================================== --- branches/Branch_2_2_EAP/src/main/org/hornetq/core/server/cluster/impl/ClusterManagerImpl.java 2011-09-19 23:47:54 UTC (rev 11370) +++ branches/Branch_2_2_EAP/src/main/org/hornetq/core/server/cluster/impl/ClusterManagerImpl.java 2011-09-20 00:50:32 UTC (rev 11371) @@ -30,15 +30,10 @@ import java.util.concurrent.ScheduledFuture; import org.hornetq.api.core.DiscoveryGroupConfiguration; -import org.hornetq.api.core.Pair; import org.hornetq.api.core.SimpleString; import org.hornetq.api.core.TransportConfiguration; -import org.hornetq.api.core.client.ClientSessionFactory; -import org.hornetq.api.core.client.ClusterTopologyListener; import org.hornetq.api.core.client.HornetQClient; import org.hornetq.core.client.impl.ServerLocatorInternal; -import org.hornetq.core.client.impl.Topology; -import org.hornetq.core.client.impl.TopologyMember; import org.hornetq.core.config.BridgeConfiguration; import org.hornetq.core.config.BroadcastGroupConfiguration; import org.hornetq.core.config.ClusterConnectionConfiguration; @@ -46,7 +41,6 @@ import org.hornetq.core.logging.Logger; import org.hornetq.core.postoffice.Binding; import org.hornetq.core.postoffice.PostOffice; -import org.hornetq.core.protocol.core.impl.wireformat.NodeAnnounceMessage; import org.hornetq.core.server.HornetQServer; import org.hornetq.core.server.Queue; import org.hornetq.core.server.cluster.Bridge; @@ -83,6 +77,8 @@ private final PostOffice postOffice; private final ScheduledExecutorService scheduledExecutor; + + private ClusterConnection defaultClusterConnection; private final ManagementService managementService; @@ -99,10 +95,6 @@ // the cluster connections which links this node to other cluster nodes private final Map<String, ClusterConnection> clusterConnections = new HashMap<String, ClusterConnection>(); - private final Topology topology = new Topology(this); - - private volatile ServerLocatorInternal backupServerLocator; - private final Set<ServerLocatorInternal> clusterLocators = new ConcurrentHashSet<ServerLocatorInternal>(); private final Executor executor; @@ -126,8 +118,6 @@ executor = executorFactory.getExecutor();; - topology.setExecutor(executor); - this.server = server; this.postOffice = postOffice; @@ -152,7 +142,6 @@ out.println("Information on " + this); out.println("*******************************************************"); - out.println("Topology: " + topology.describe("Toopology on " + this)); for (ClusterConnection conn : this.clusterConnections.values()) { @@ -163,29 +152,24 @@ return str.toString(); } + + public ClusterConnection getDefaultConnection() + { + return defaultClusterConnection; + } public String toString() { return "ClusterManagerImpl[server=" + server + "]@" + System.identityHashCode(this); } - public TopologyMember getLocalMember() - { - return topology.getMember(nodeUUID.toString()); - } - public String getNodeId() { return nodeUUID.toString(); } - public synchronized void start() throws Exception + public synchronized void deploy() throws Exception { - if (started) - { - return; - } - if (clustered) { for (BroadcastGroupConfiguration config : configuration.getBroadcastGroupConfigurations()) @@ -193,43 +177,44 @@ deployBroadcastGroup(config); } - String connectorName = null; - for (ClusterConnectionConfiguration config : configuration.getClusterConfigurations()) { - if (connectorName == null) - { - connectorName = config.getConnectorName(); - break; - } - } + deployClusterConnection(config); + } + } + } - if (connectorName != null) + public synchronized void start() throws Exception + { + if (started) + { + return; + } + + for (BroadcastGroup group: broadcastGroups.values()) + { + if (!backup) { - TransportConfiguration nodeConnector = configuration.getConnectorConfigurations().get(connectorName); - if (nodeConnector == null) - { - log.warn("No connecor with name '" + connectorName + - "'. The cluster connection will not be deployed."); - return; - } - - // Now announce presence - announceNode(nodeConnector); - - for (ClusterConnectionConfiguration config : configuration.getClusterConfigurations()) - { - deployClusterConnection(config); - } + group.start(); } - } + + for (ClusterConnection conn : clusterConnections.values()) + { + conn.start(); + if (backup) + { + conn.informTopology(); + conn.announceBackup(); + } + } for (BridgeConfiguration config : configuration.getBridgeConfigurations()) { - deployBridge(config); + deployBridge(config, !backup); } + started = true; } @@ -267,12 +252,6 @@ } bridges.clear(); - - if (backupServerLocator != null) - { - backupServerLocator.close(); - backupServerLocator = null; - } } for (ServerLocatorInternal clusterLocator : clusterLocators) @@ -289,31 +268,9 @@ clusterLocators.clear(); started = false; - clusterConnections.clear(); + clearClusterConnections(); } - public void nodeAnnounced(final long uniqueEventID, - final String nodeID, - final Pair<TransportConfiguration, TransportConfiguration> connectorPair, - final boolean backup) - { - if (log.isDebugEnabled()) - { - log.debug(this + "::NodeAnnounced, backup=" + backup + nodeID + connectorPair); - } - - TopologyMember newMember = new TopologyMember(connectorPair.a, connectorPair.b); - newMember.setUniqueEventID(uniqueEventID); - if (backup) - { - topology.updateBackup(nodeID, new TopologyMember(connectorPair.a, connectorPair.b)); - } - else - { - topology.updateMember(uniqueEventID, nodeID, newMember); - } - } - public void flushExecutor() { Future future = new Future(); @@ -350,24 +307,6 @@ return clusterConnections.get(name.toString()); } - public void addClusterTopologyListener(final ClusterTopologyListener listener, final boolean clusterConnection) - { - topology.addClusterTopologyListener(listener); - - // no need to use an executor here since the Topology is already using one - topology.sendTopology(listener); - } - - public void removeClusterTopologyListener(final ClusterTopologyListener listener, final boolean clusterConnection) - { - topology.removeClusterTopologyListener(listener); - } - - public Topology getTopology() - { - return topology; - } - // backup node becomes live public synchronized void activate() { @@ -375,27 +314,6 @@ { backup = false; - String nodeID = server.getNodeID().toString(); - - TopologyMember member = topology.getMember(nodeID); - // swap backup as live and send it to everybody - member = new TopologyMember(member.getConnector().b, null); - topology.updateAsLive(nodeID, member); - - if (backupServerLocator != null) - { - // todo we could use the topology of this to preempt it arriving from the cc - try - { - backupServerLocator.close(); - } - catch (Exception e) - { - log.warn("problem closing backup session factory", e); - } - backupServerLocator = null; - } - for (BroadcastGroup broadcastGroup : broadcastGroups.values()) { try @@ -432,31 +350,15 @@ log.warn("unable to start bridge " + bridge.getName(), e); } } - - topology.sendMember(nodeID); } } public void announceBackup() throws Exception { - List<ClusterConnectionConfiguration> configs = this.configuration.getClusterConfigurations(); - if (!configs.isEmpty()) + for (ClusterConnection conn : this.clusterConnections.values()) { - ClusterConnectionConfiguration config = configs.get(0); - - TransportConfiguration connector = configuration.getConnectorConfigurations().get(config.getConnectorName()); - - if (connector == null) - { - log.warn("No connecor with name '" + config.getConnectorName() + "'. backup cannot be announced."); - return; - } - announceBackup(config, connector); + conn.announceBackup(); } - else - { - log.warn("no cluster connections defined, unable to announce backup"); - } } public void addClusterLocator(final ServerLocatorInternal serverLocator) @@ -468,114 +370,9 @@ { this.clusterLocators.remove(serverLocator); } - - private synchronized void announceNode(final TransportConfiguration nodeConnector) + + public synchronized void deployBridge(final BridgeConfiguration config, final boolean start) throws Exception { - String nodeID = server.getNodeID().toString(); - - TopologyMember localMember; - if (backup) - { - localMember = new TopologyMember(null, nodeConnector); - } - else - { - localMember = new TopologyMember(nodeConnector, null); - } - - topology.updateAsLive(nodeID, localMember); - } - - private synchronized void deployBroadcastGroup(final BroadcastGroupConfiguration config) throws Exception - { - if (broadcastGroups.containsKey(config.getName())) - { - ClusterManagerImpl.log.warn("There is already a broadcast-group with name " + config.getName() + - " deployed. This one will not be deployed."); - - return; - } - - InetAddress localAddress = null; - if (config.getLocalBindAddress() != null) - { - localAddress = InetAddress.getByName(config.getLocalBindAddress()); - } - - InetAddress groupAddress = InetAddress.getByName(config.getGroupAddress()); - - BroadcastGroupImpl group = new BroadcastGroupImpl(nodeUUID.toString(), - config.getName(), - localAddress, - config.getLocalBindPort(), - groupAddress, - config.getGroupPort(), - !backup); - - for (String connectorInfo : config.getConnectorInfos()) - { - TransportConfiguration connector = configuration.getConnectorConfigurations().get(connectorInfo); - - if (connector == null) - { - logWarnNoConnector(config.getName(), connectorInfo); - - return; - } - - group.addConnector(connector); - } - - ScheduledFuture<?> future = scheduledExecutor.scheduleWithFixedDelay(group, - 0L, - config.getBroadcastPeriod(), - MILLISECONDS); - - group.setScheduledFuture(future); - - broadcastGroups.put(config.getName(), group); - - managementService.registerBroadcastGroup(group, config); - - if (!backup) - { - group.start(); - } - } - - private void logWarnNoConnector(final String connectorName, final String bgName) - { - ClusterManagerImpl.log.warn("There is no connector deployed with name '" + connectorName + - "'. The broadcast group with name '" + - bgName + - "' will not be deployed."); - } - - private TransportConfiguration[] connectorNameListToArray(final List<String> connectorNames) - { - TransportConfiguration[] tcConfigs = (TransportConfiguration[])Array.newInstance(TransportConfiguration.class, - connectorNames.size()); - int count = 0; - for (String connectorName : connectorNames) - { - TransportConfiguration connector = configuration.getConnectorConfigurations().get(connectorName); - - if (connector == null) - { - ClusterManagerImpl.log.warn("No connector defined with name '" + connectorName + - "'. The bridge will not be deployed."); - - return null; - } - - tcConfigs[count++] = connector; - } - - return tcConfigs; - } - - public synchronized void deployBridge(final BridgeConfiguration config) throws Exception - { if (config.getName() == null) { ClusterManagerImpl.log.warn("Must specify a unique name for each bridge. This one will not be deployed."); @@ -702,11 +499,12 @@ bridges.put(config.getName(), bridge); managementService.registerBridge(bridge, config); - - if (!backup) + + if (start) { bridge.start(); } + } public void destroyBridge(final String name) throws Exception @@ -726,11 +524,49 @@ bridge.flushExecutor(); } - private synchronized void deployClusterConnection(final ClusterConnectionConfiguration config) throws Exception + // for testing + public void clear() { + for (Bridge bridge : bridges.values()) + { + try + { + bridge.stop(); + } + catch (Exception e) + { + log.warn(e.getMessage(), e); + } + } + bridges.clear(); + for (ClusterConnection clusterConnection : clusterConnections.values()) + { + try + { + clusterConnection.stop(); + } + catch (Exception e) + { + e.printStackTrace(); + } + } + clearClusterConnections(); + } + + // Private methods ---------------------------------------------------------------------------------------------------- + + + private void clearClusterConnections() + { + clusterConnections.clear(); + this.defaultClusterConnection = null; + } + + private void deployClusterConnection(final ClusterConnectionConfiguration config) throws Exception + { if (config.getName() == null) { - ClusterManagerImpl.log.warn("Must specify a unique name for each cluster. This one will not be deployed."); + ClusterManagerImpl.log.warn("Must specify a unique name for each cluster connection. This one will not be deployed."); return; } @@ -781,7 +617,6 @@ } clusterConnection = new ClusterConnectionImpl(this, - topology, dg, connector, new SimpleString(config.getName()), @@ -819,7 +654,6 @@ } clusterConnection = new ClusterConnectionImpl(this, - topology, tcConfigs, connector, new SimpleString(config.getName()), @@ -847,6 +681,11 @@ config.isAllowDirectConnectionsOnly()); } + if (defaultClusterConnection == null) + { + defaultClusterConnection = clusterConnection; + } + managementService.registerCluster(clusterConnection, config); clusterConnections.put(config.getName(), clusterConnection); @@ -855,75 +694,8 @@ { log.debug("ClusterConnection.start at " + clusterConnection, new Exception("trace")); } - clusterConnection.start(); - - if (backup) - { - announceBackup(config, connector); - } } - - private void announceBackup(final ClusterConnectionConfiguration config, final TransportConfiguration connector) throws Exception - { - if (config.getStaticConnectors() != null) - { - TransportConfiguration[] tcConfigs = connectorNameListToArray(config.getStaticConnectors()); - - backupServerLocator = (ServerLocatorInternal)HornetQClient.createServerLocatorWithoutHA(tcConfigs); - backupServerLocator.setReconnectAttempts(-1); - backupServerLocator.setInitialConnectAttempts(-1); - } - else if (config.getDiscoveryGroupName() != null) - { - DiscoveryGroupConfiguration dg = configuration.getDiscoveryGroupConfigurations() - .get(config.getDiscoveryGroupName()); - - if (dg == null) - { - ClusterManagerImpl.log.warn("No discovery group with name '" + config.getDiscoveryGroupName() + - "'. The cluster connection will not be deployed."); - } - - backupServerLocator = (ServerLocatorInternal)HornetQClient.createServerLocatorWithoutHA(dg); - backupServerLocator.setReconnectAttempts(-1); - backupServerLocator.setInitialConnectAttempts(-1); - } - else - { - return; - } - log.info("announcing backup"); - executor.execute(new Runnable() - { - public void run() - { - try - { - if (log.isDebugEnabled()) - { - log.debug(ClusterManagerImpl.this + ":: announcing " + connector + " to " + backupServerLocator); - } - ClientSessionFactory backupSessionFactory = backupServerLocator.connect(); - if (backupSessionFactory != null) - { - backupSessionFactory.getConnection() - .getChannel(0, -1) - .send(new NodeAnnounceMessage(System.currentTimeMillis(), - nodeUUID.toString(), - true, - connector, - null)); - log.info("backup announced"); - } - } - catch (Exception e) - { - log.warn("Unable to announce backup, retrying", e); - } - } - }); - } - + private Transformer instantiateTransformer(final String transformerClassName) { Transformer transformer = null; @@ -945,32 +717,89 @@ return transformer; } - // for testing - public void clear() + + private synchronized void deployBroadcastGroup(final BroadcastGroupConfiguration config) throws Exception { - for (Bridge bridge : bridges.values()) + if (broadcastGroups.containsKey(config.getName())) { - try + ClusterManagerImpl.log.warn("There is already a broadcast-group with name " + config.getName() + + " deployed. This one will not be deployed."); + + return; + } + + InetAddress localAddress = null; + if (config.getLocalBindAddress() != null) + { + localAddress = InetAddress.getByName(config.getLocalBindAddress()); + } + + InetAddress groupAddress = InetAddress.getByName(config.getGroupAddress()); + + BroadcastGroupImpl group = new BroadcastGroupImpl(nodeUUID.toString(), + config.getName(), + localAddress, + config.getLocalBindPort(), + groupAddress, + config.getGroupPort(), + !backup); + + for (String connectorInfo : config.getConnectorInfos()) + { + TransportConfiguration connector = configuration.getConnectorConfigurations().get(connectorInfo); + + if (connector == null) { - bridge.stop(); + logWarnNoConnector(config.getName(), connectorInfo); + + return; } - catch (Exception e) - { - log.warn(e.getMessage(), e); - } + + group.addConnector(connector); } - bridges.clear(); - for (ClusterConnection clusterConnection : clusterConnections.values()) + + ScheduledFuture<?> future = scheduledExecutor.scheduleWithFixedDelay(group, + 0L, + config.getBroadcastPeriod(), + MILLISECONDS); + + group.setScheduledFuture(future); + + broadcastGroups.put(config.getName(), group); + + managementService.registerBroadcastGroup(group, config); + } + + private void logWarnNoConnector(final String connectorName, final String bgName) + { + ClusterManagerImpl.log.warn("There is no connector deployed with name '" + connectorName + + "'. The broadcast group with name '" + + bgName + + "' will not be deployed."); + } + + private TransportConfiguration[] connectorNameListToArray(final List<String> connectorNames) + { + TransportConfiguration[] tcConfigs = (TransportConfiguration[])Array.newInstance(TransportConfiguration.class, + connectorNames.size()); + int count = 0; + for (String connectorName : connectorNames) { - try + TransportConfiguration connector = configuration.getConnectorConfigurations().get(connectorName); + + if (connector == null) { - clusterConnection.stop(); + ClusterManagerImpl.log.warn("No connector defined with name '" + connectorName + + "'. The bridge will not be deployed."); + + return null; } - catch (Exception e) - { - e.printStackTrace(); - } + + tcConfigs[count++] = connector; } - clusterConnections.clear(); + + return tcConfigs; } + + } Modified: branches/Branch_2_2_EAP/src/main/org/hornetq/core/server/cluster/impl/ClusterManagerInternal.java =================================================================== --- branches/Branch_2_2_EAP/src/main/org/hornetq/core/server/cluster/impl/ClusterManagerInternal.java 2011-09-19 23:47:54 UTC (rev 11370) +++ branches/Branch_2_2_EAP/src/main/org/hornetq/core/server/cluster/impl/ClusterManagerInternal.java 2011-09-20 00:50:32 UTC (rev 11371) @@ -14,7 +14,6 @@ package org.hornetq.core.server.cluster.impl; import org.hornetq.core.client.impl.ServerLocatorInternal; -import org.hornetq.core.client.impl.TopologyMember; import org.hornetq.core.server.cluster.ClusterManager; /** @@ -30,8 +29,6 @@ void removeClusterLocator(ServerLocatorInternal locator); - TopologyMember getLocalMember(); - String getNodeId(); } Modified: branches/Branch_2_2_EAP/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java =================================================================== --- branches/Branch_2_2_EAP/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java 2011-09-19 23:47:54 UTC (rev 11370) +++ branches/Branch_2_2_EAP/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java 2011-09-20 00:50:32 UTC (rev 11371) @@ -227,7 +227,12 @@ // Used to identify the server on tests... useful on debugging testcases private String identity; + + private Thread backupActivationThread; + private Activation activation; + + // Constructors // --------------------------------------------------------------------------------- @@ -289,11 +294,6 @@ // lifecycle methods // ---------------------------------------------------------------- - private interface Activation extends Runnable - { - void close(boolean permanently) throws Exception; - } - /* * Can be overridden for tests */ @@ -309,259 +309,6 @@ } } - private class NoSharedStoreLiveActivation implements Activation - { - public void run() - { - try - { - initialisePart1(); - - initialisePart2(); - - if (identity != null) - { - log.info("Server " + identity + " is now live"); - } - else - { - log.info("Server is now live"); - } - } - catch (Exception e) - { - log.error("Failure in initialisation", e); - } - } - - public void close(boolean permanently) throws Exception - { - - } - } - - private class SharedStoreLiveActivation implements Activation - { - public void run() - { - try - { - log.info("Waiting to obtain live lock"); - - checkJournalDirectory(); - - initialisePart1(); - - if(nodeManager.isBackupLive()) - { - //looks like we've failed over at some point need to inform that we are the backup so when the current live - // goes down they failover to us - clusterManager.announceBackup(); - Thread.sleep(configuration.getFailbackDelay()); - } - - nodeManager.startLiveNode(); - - if (stopped) - { - return; - } - - initialisePart2(); - - log.info("Server is now live"); - } - catch (Exception e) - { - log.error("Failure in initialisation", e); - } - } - - public void close(boolean permanently) throws Exception - { - if(permanently) - { - nodeManager.crashLiveServer(); - } - else - { - nodeManager.pauseLiveServer(); - } - } - } - - - private class SharedStoreBackupActivation implements Activation - { - - volatile boolean closed = false; - public void run() - { - try - { - nodeManager.startBackup(); - - initialisePart1(); - - clusterManager.start(); - - started = true; - - log.info("HornetQ Backup Server version " + getVersion().getFullVersion() + " [" + nodeManager.getNodeId() + "] started, waiting live to fail before it gets active"); - - nodeManager.awaitLiveNode(); - - configuration.setBackup(false); - - if (stopped) - { - return; - } - - initialisePart2(); - - clusterManager.activate(); - - log.info("Backup Server is now live"); - - nodeManager.releaseBackup(); - if(configuration.isAllowAutoFailBack()) - { - class FailbackChecker implements Runnable - { - boolean restarting = false; - public void run() - { - try - { - if(!restarting && nodeManager.isAwaitingFailback()) - { - log.info("live server wants to restart, restarting server in backup"); - restarting = true; - Thread t = new Thread(new Runnable() - { - public void run() - { - try - { - log.debug(HornetQServerImpl.this + "::Stopping live node in favor of failback"); - stop(true); - // We need to wait some time before we start the backup again - // otherwise we may eventually start before the live had a chance to get it - Thread.sleep(configuration.getFailbackDelay()); - configuration.setBackup(true); - log.debug(HornetQServerImpl.this + "::Starting backup node now after failback"); - start(); - } - catch (Exception e) - { - log.warn("unable to restart server, please kill and restart manually", e); - } - } - }); - t.start(); - } - } - catch (Exception e) - { - log.debug(e.getMessage(), e); - //hopefully it will work next call - } - } - } - scheduledPool.scheduleAtFixedRate(new FailbackChecker(), 1000l, 1000l, TimeUnit.MILLISECONDS); - } - } - catch (InterruptedException e) - { - //this is ok, we are being stopped - } - catch (ClosedChannelException e) - { - //this is ok too, we are being stopped - } - catch (Exception e) - { - if(!(e.getCause() instanceof InterruptedException)) - { - log.error("Failure in initialisation", e); - } - } - catch(Throwable e) - { - log.error("Failure in initialisation", e); - } - } - - public void close(boolean permanently) throws Exception - { - if (configuration.isBackup()) - { - long timeout = 30000; - - long start = System.currentTimeMillis(); - - while (backupActivationThread.isAlive() && System.currentTimeMillis() - start < timeout) - { - nodeManager.interrupt(); - - backupActivationThread.interrupt(); - - backupActivationThread.join(1000); - - } - - if (System.currentTimeMillis() - start >= timeout) - { - threadDump("Timed out waiting for backup activation to exit"); - } - - nodeManager.stopBackup(); - } - else - { - //if we are now live, behave as live - // We need to delete the file too, otherwise the backup will failover when we shutdown or if the backup is - // started before the live - if(permanently) - { - nodeManager.crashLiveServer(); - } - else - { - nodeManager.pauseLiveServer(); - } - } - } - } - - private class SharedNothingBackupActivation implements Activation - { - public void run() - { - try - { - // TODO - - // Try-Connect to live server using live-connector-ref - - // sit in loop and try and connect, if server is not live then it will return NOT_LIVE - } - catch (Exception e) - { - log.error("Failure in initialisation", e); - } - } - - public void close(boolean permanently) throws Exception - { - } - } - - private Thread backupActivationThread; - - private Activation activation; - public synchronized void start() throws Exception { stopped = false; @@ -611,6 +358,7 @@ } + // The activation on fail-back may change the value of isBackup, for that reason we are not using else here if (configuration.isBackup()) { if (configuration.isSharedStore()) @@ -1069,7 +817,6 @@ return new HashSet<ServerSession>(sessions.values()); } - // TODO - should this really be here?? It's only used in tests public boolean isInitialised() { synchronized (initialiseLock) @@ -1232,9 +979,145 @@ return connectorsService; } - // Public - // --------------------------------------------------------------------------------------- + + public synchronized boolean checkActivate() throws Exception + { + if (configuration.isBackup()) + { + // Handle backup server activation + if (!configuration.isSharedStore()) + { + if (replicationEndpoint == null) + { + HornetQServerImpl.log.warn("There is no replication endpoint, can't activate this backup server"); + + throw new HornetQException(HornetQException.INTERNAL_ERROR, "Can't activate the server"); + } + + replicationEndpoint.stop(); + } + + // Complete the startup procedure + + HornetQServerImpl.log.info("Activating backup server"); + + configuration.setBackup(false); + + initialisePart2(); + } + + return true; + } + + public void deployDivert(DivertConfiguration config) throws Exception + { + if (config.getName() == null) + { + HornetQServerImpl.log.warn("Must specify a name for each divert. This one will not be deployed."); + + return; + } + + if (config.getAddress() == null) + { + HornetQServerImpl.log.warn("Must specify an address for each divert. This one will not be deployed."); + + return; + } + + if (config.getForwardingAddress() == null) + { + HornetQServerImpl.log.warn("Must specify an forwarding address for each divert. This one will not be deployed."); + + return; + } + + SimpleString sName = new SimpleString(config.getName()); + + if (postOffice.getBinding(sName) != null) + { + HornetQServerImpl.log.warn("Binding already exists with name " + sName + ", divert will not be deployed"); + + return; + } + + SimpleString sAddress = new SimpleString(config.getAddress()); + + Transformer transformer = instantiateTransformer(config.getTransformerClassName()); + + Filter filter = FilterImpl.createFilter(config.getFilterString()); + + Divert divert = new DivertImpl(new SimpleString(config.getForwardingAddress()), + sName, + new SimpleString(config.getRoutingName()), + config.isExclusive(), + filter, + transformer, + postOffice, + storageManager); + + Binding binding = new DivertBinding(storageManager.generateUniqueID(), sAddress, divert); + + postOffice.addBinding(binding); + + managementService.registerDivert(divert, config); + } + + public void destroyDivert(SimpleString name) throws Exception + { + Binding binding = postOffice.getBinding(name); + if (binding == null) + { + throw new HornetQException(HornetQException.INTERNAL_ERROR, "No binding for divert " + name); + } + if (!(binding instanceof DivertBinding)) + { + throw new HornetQException(HornetQException.INTERNAL_ERROR, "Binding " + name + " is not a divert"); + } + + postOffice.removeBinding(name); + } + + + + public void deployBridge(BridgeConfiguration config) throws Exception + { + if (clusterManager != null) + { + clusterManager.deployBridge(config, true); + } + } + + public void destroyBridge(String name) throws Exception + { + if (clusterManager != null) + { + clusterManager.destroyBridge(name); + } + } + + public ServerSession getSessionByID(String sessionName) + { + return sessions.get(sessionName); + } + + // PUBLIC ------- + + public String toString() + { + if (identity != null) + { + return "HornetQServerImpl::" + identity; + } + else + { + return "HornetQServerImpl::" + (nodeManager != null ? "serverUUID=" + nodeManager.getUUID() : ""); + } + } + + + // Package protected // ---------------------------------------------------------------------------- @@ -1296,34 +1179,6 @@ // Private // -------------------------------------------------------------------------------------- - // private boolean startReplication() throws Exception - // { - // String backupConnectorName = configuration.getBackupConnectorName(); - // - // if (!configuration.isSharedStore() && backupConnectorName != null) - // { - // TransportConfiguration backupConnector = configuration.getConnectorConfigurations().get(backupConnectorName); - // - // if (backupConnector == null) - // { - // HornetQServerImpl.log.warn("connector with name '" + backupConnectorName + - // "' is not defined in the configuration."); - // } - // else - // { - // - // replicationFailoverManager = createBackupConnectionFailoverManager(backupConnector, - // threadPool, - // scheduledPool); - // - // replicationManager = new ReplicationManagerImpl(replicationFailoverManager, executorFactory); - // replicationManager.start(); - // } - // } - // - // return true; - // } - private void callActivateCallbacks() { for (ActivateCallback callback : activateCallbacks) @@ -1340,44 +1195,6 @@ } } - public synchronized boolean checkActivate() throws Exception - { - if (configuration.isBackup()) - { - // Handle backup server activation - - if (!configuration.isSharedStore()) - { - if (replicationEndpoint == null) - { - HornetQServerImpl.log.warn("There is no replication endpoint, can't activate this backup server"); - - throw new HornetQException(HornetQException.INTERNAL_ERROR, "Can't activate the server"); - } - - replicationEndpoint.stop(); - } - - // Complete the startup procedure - - HornetQServerImpl.log.info("Activating backup server"); - - configuration.setBackup(false); - - initialisePart2(); - } - - return true; - } - - private class FileActivateRunner implements Runnable - { - public void run() - { - - } - } - private void initialiseLogging() { LogDelegateFactory logDelegateFactory = (LogDelegateFactory)instantiateInstance(configuration.getLogDelegateFactoryClassName()); @@ -1414,8 +1231,6 @@ managementService = new ManagementServiceImpl(mbeanServer, configuration); - remotingService = new RemotingServiceImpl(configuration, this, managementService, scheduledPool); - if (configuration.getMemoryMeasureInterval() != -1) { memoryManager = new MemoryManagerImpl(configuration.getMemoryWarningThreshold(), @@ -1470,6 +1285,23 @@ configuration.isPersistIDCache(), addressSettingsRepository); + // This can't be created until node id is set + clusterManager = new ClusterManagerImpl(executorFactory, + this, + postOffice, + scheduledPool, + managementService, + configuration, + nodeManager.getUUID(), + configuration.isBackup(), + configuration.isClustered()); + + + clusterManager.deploy(); + + + remotingService = new RemotingServiceImpl(clusterManager, configuration, this, managementService, scheduledPool); + messagingServerControl = managementService.registerServer(postOffice, storageManager, configuration, @@ -1527,18 +1359,6 @@ deploySecurityFromConfiguration(); deployGroupingHandlerConfiguration(configuration.getGroupingHandlerConfiguration()); - - // This can't be created until node id is set - clusterManager = new ClusterManagerImpl(executorFactory, - this, - postOffice, - scheduledPool, - managementService, - configuration, - nodeManager.getUUID(), - configuration.isBackup(), - configuration.isClustered()); - } /* @@ -1604,10 +1424,10 @@ // We do this at the end - we don't want things like MDBs or other connections connecting to a backup server until // it is activated - remotingService.start(); - clusterManager.start(); + remotingService.start(); + initialised = true; } @@ -1826,76 +1646,6 @@ } } - public void deployDivert(DivertConfiguration config) throws Exception - { - if (config.getName() == null) - { - HornetQServerImpl.log.warn("Must specify a name for each divert. This one will not be deployed."); - - return; - } - - if (config.getAddress() == null) - { - HornetQServerImpl.log.warn("Must specify an address for each divert. This one will not be deployed."); - - return; - } - - if (config.getForwardingAddress() == null) - { - HornetQServerImpl.log.warn("Must specify an forwarding address for each divert. This one will not be deployed."); - - return; - } - - SimpleString sName = new SimpleString(config.getName()); - - if (postOffice.getBinding(sName) != null) - { - HornetQServerImpl.log.warn("Binding already exists with name " + sName + ", divert will not be deployed"); - - return; - } - - SimpleString sAddress = new SimpleString(config.getAddress()); - - Transformer transformer = instantiateTransformer(config.getTransformerClassName()); - - Filter filter = FilterImpl.createFilter(config.getFilterString()); - - Divert divert = new DivertImpl(new SimpleString(config.getForwardingAddress()), - sName, - new SimpleString(config.getRoutingName()), - config.isExclusive(), - filter, - transformer, - postOffice, - storageManager); - - Binding binding = new DivertBinding(storageManager.generateUniqueID(), sAddress, divert); - - postOffice.addBinding(binding); - - managementService.registerDivert(divert, config); - } - - public void destroyDivert(SimpleString name) throws Exception - { - Binding binding = postOffice.getBinding(name); - if (binding == null) - { - throw new HornetQException(HornetQException.INTERNAL_ERROR, "No binding for divert " + name); - } - if (!(binding instanceof DivertBinding)) - { - throw new HornetQException(HornetQException.INTERNAL_ERROR, "Binding " + name + " is not a divert"); - } - - postOffice.removeBinding(name); - } - - private synchronized void deployGroupingHandlerConfiguration(final GroupingHandlerConfiguration config) throws Exception { if (config != null) @@ -1922,22 +1672,6 @@ managementService.addNotificationListener(groupingHandler); } } - - public void deployBridge(BridgeConfiguration config) throws Exception - { - if (clusterManager != null) - { - clusterManager.deployBridge(config); - } - } - - public void destroyBridge(String name) throws Exception - { - if (clusterManager != null) - { - clusterManager.destroyBridge(name); - } - } private Transformer instantiateTransformer(final String transformerClassName) { @@ -1979,11 +1713,6 @@ } - public ServerSession getSessionByID(String sessionName) - { - return sessions.get(sessionName); - } - /** * Check if journal directory exists or create it (if configured to do so) */ @@ -2005,18 +1734,284 @@ } } - public String toString() + /** + * To be called by backup trying to fail back the server + */ + private void startFailbackChecker() { - if (identity != null) + scheduledPool.scheduleAtFixedRate(new FailbackChecker(), 1000l, 1000l, TimeUnit.MILLISECONDS); + } + + + // Inner classes + // -------------------------------------------------------------------------------- + + class FailbackChecker implements Runnable + { + boolean restarting = false; + public void run() { - return "HornetQServerImpl::" + identity; + try + { + if(!restarting && nodeManager.isAwaitingFailback()) + { + log.info("live server wants to restart, restarting server in backup"); + restarting = true; + Thread t = new Thread(new Runnable() + { + public void run() + { + try + { + log.debug(HornetQServerImpl.this + "::Stopping live node in favor of failback"); + stop(true); + // We need to wait some time before we start the backup again + // otherwise we may eventually start before the live had a chance to get it + Thread.sleep(configuration.getFailbackDelay()); + configuration.setBackup(true); + log.debug(HornetQServerImpl.this + "::Starting backup node now after failback"); + start(); + } + catch (Exception e) + { + log.warn("unable to restart server, please kill and restart manually", e); + } + } + }); + t.start(); + } + } + catch (Exception e) + { + log.warn(e.getMessage(), e); + } } - else + } + + + + private class SharedStoreLiveActivation implements Activation + { + public void run() { - return "HornetQServerImpl::" + (nodeManager != null ? "serverUUID=" + nodeManager.getUUID() : ""); + try + { + log.info("Waiting to obtain live lock"); + + checkJournalDirectory(); + + if (log.isDebugEnabled()) + { + log.debug("First part initialization on " + this); + } + + initialisePart1(); + + if(nodeManager.isBackupLive()) + { + //looks like we've failed over at some point need to inform that we are the backup so when the current live + // goes down they failover to us + if (log.isDebugEnabled()) + { + log.debug("announcing backup to the former live" + this); + } + + clusterManager.announceBackup(); + Thread.sleep(configuration.getFailbackDelay()); + } + + nodeManager.startLiveNode(); + + if (stopped) + { + return; + } + + initialisePart2(); + + log.info("Server is now live"); + } + catch (Exception e) + { + log.error("Failure in initialisation", e); + } } + + public void close(boolean permanently) throws Exception + { + if(permanently) + { + nodeManager.crashLiveServer(); + } + else + { + nodeManager.pauseLiveServer(); + } + } } - // Inner classes - // -------------------------------------------------------------------------------- + + private class SharedStoreBackupActivation implements Activation + { + public void run() + { + try + { + nodeManager.startBackup(); + + initialisePart1(); + + clusterManager.start(); + + started = true; + + log.info("HornetQ Backup Server version " + getVersion().getFullVersion() + " [" + nodeManager.getNodeId() + "] started, waiting live to fail before it gets active"); + + nodeManager.awaitLiveNode(); + + configuration.setBackup(false); + + if (stopped) + { + return; + } + + initialisePart2(); + + clusterManager.activate(); + + log.info("Backup Server is now live"); + + nodeManager.releaseBackup(); + if(configuration.isAllowAutoFailBack()) + { + startFailbackChecker(); + } + } + catch (InterruptedException e) + { + //this is ok, we are being stopped + } + catch (ClosedChannelException e) + { + //this is ok too, we are being stopped + } + catch (Exception e) + { + if(!(e.getCause() instanceof InterruptedException)) + { + log.error("Failure in initialisation", e); + } + } + catch(Throwable e) + { + log.error("Failure in initialisation", e); + } + } + + /** + * + */ + public void close(boolean permanently) throws Exception + { + if (configuration.isBackup()) + { + long timeout = 30000; + + long start = System.currentTimeMillis(); + + while (backupActivationThread.isAlive() && System.currentTimeMillis() - start < timeout) + { + nodeManager.interrupt(); + + backupActivationThread.interrupt(); + + backupActivationThread.join(1000); + + } + + if (System.currentTimeMillis() - start >= timeout) + { + threadDump("Timed out waiting for backup activation to exit"); + } + + nodeManager.stopBackup(); + } + else + { + //if we are now live, behave as live + // We need to delete the file too, otherwise the backup will failover when we shutdown or if the backup is + // started before the live + if(permanently) + { + nodeManager.crashLiveServer(); + } + else + { + nodeManager.pauseLiveServer(); + } + } + } + } + + private interface Activation extends Runnable + { + void close(boolean permanently) throws Exception; + } + + private class SharedNothingBackupActivation implements Activation + { + public void run() + { + try + { + // TODO + + // Try-Connect to live server using live-connector-ref + + // sit in loop and try and connect, if server is not live then it will return NOT_LIVE + } + catch (Exception e) + { + log.error("Failure in initialisation", e); + } + } + + public void close(boolean permanently) throws Exception + { + } + } + + private class NoSharedStoreLiveActivation implements Activation + { + public void run() + { + try + { + initialisePart1(); + + initialisePart2(); + + if (identity != null) + { + log.info("Server " + identity + " is now live"); + } + else + { + log.info("Server is now live"); + } + } + catch (Exception e) + { + log.error("Failure in initialisation", e); + } + } + + public void close(boolean permanently) throws Exception + { + + } + } + + } Modified: branches/Branch_2_2_EAP/src/main/org/hornetq/core/server/impl/ServerSessionImpl.java =================================================================== --- branches/Branch_2_2_EAP/src/main/org/hornetq/core/server/impl/ServerSessionImpl.java 2011-09-19 23:47:54 UTC (rev 11370) +++ branches/Branch_2_2_EAP/src/main/org/hornetq/core/server/impl/ServerSessionImpl.java 2011-09-20 00:50:32 UTC (rev 11371) @@ -1062,7 +1062,7 @@ return; } - + consumer.receiveCredits(credits); } Modified: branches/Branch_2_2_EAP/src/main/org/hornetq/spi/core/protocol/ProtocolManager.java =================================================================== --- branches/Branch_2_2_EAP/src/main/org/hornetq/spi/core/protocol/ProtocolManager.java 2011-09-19 23:47:54 UTC (rev 11370) +++ branches/Branch_2_2_EAP/src/main/org/hornetq/spi/core/protocol/ProtocolManager.java 2011-09-20 00:50:32 UTC (rev 11371) @@ -14,6 +14,7 @@ package org.hornetq.spi.core.protocol; import org.hornetq.api.core.HornetQBuffer; +import org.hornetq.spi.core.remoting.Acceptor; import org.hornetq.spi.core.remoting.BufferDecoder; import org.hornetq.spi.core.remoting.Connection; @@ -26,7 +27,7 @@ */ public interface ProtocolManager extends BufferDecoder { - ConnectionEntry createConnectionEntry(Connection connection); + ConnectionEntry createConnectionEntry(Acceptor acceptorUsed, Connection connection); public void removeHandler(final String name); Modified: branches/Branch_2_2_EAP/src/main/org/hornetq/spi/core/remoting/Acceptor.java =================================================================== --- branches/Branch_2_2_EAP/src/main/org/hornetq/spi/core/remoting/Acceptor.java 2011-09-19 23:47:54 UTC (rev 11370) +++ branches/Branch_2_2_EAP/src/main/org/hornetq/spi/core/remoting/Acceptor.java 2011-09-20 00:50:32 UTC (rev 11371) @@ -14,6 +14,7 @@ package org.hornetq.spi.core.remoting; import org.hornetq.core.server.HornetQComponent; +import org.hornetq.core.server.cluster.ClusterConnection; import org.hornetq.core.server.management.NotificationService; /** @@ -31,6 +32,11 @@ void pause(); /** + * @return the cluster connection associated with this Acceptor + */ + ClusterConnection getClusterConnection(); + + /** * Set the notification service for this acceptor to use. * * @param notificationService the notification service Modified: branches/Branch_2_2_EAP/src/main/org/hornetq/spi/core/remoting/AcceptorFactory.java =================================================================== --- branches/Branch_2_2_EAP/src/main/org/hornetq/spi/core/remoting/AcceptorFactory.java 2011-09-19 23:47:54 UTC (rev 11370) +++ branches/Branch_2_2_EAP/src/main/org/hornetq/spi/core/remoting/AcceptorFactory.java 2011-09-20 00:50:32 UTC (rev 11371) @@ -18,6 +18,8 @@ import java.util.concurrent.Executor; import java.util.concurrent.ScheduledExecutorService; +import org.hornetq.core.server.cluster.ClusterConnection; + /** * A factory for creating acceptors. * <p/> @@ -40,7 +42,8 @@ * @param scheduledThreadPool a scheduled thread pool * @return an acceptor */ - Acceptor createAcceptor(final Map<String, Object> configuration, + Acceptor createAcceptor(ClusterConnection clusterConnection, + final Map<String, Object> configuration, BufferHandler handler, BufferDecoder decoder, ConnectionLifeCycleListener listener, Modified: branches/Branch_2_2_EAP/src/main/org/hornetq/spi/core/remoting/ConnectionLifeCycleListener.java =================================================================== --- branches/Branch_2_2_EAP/src/main/org/hornetq/spi/core/remoting/ConnectionLifeCycleListener.java 2011-09-19 23:47:54 UTC (rev 11370) +++ branches/Branch_2_2_EAP/src/main/org/hornetq/spi/core/remoting/ConnectionLifeCycleListener.java 2011-09-20 00:50:32 UTC (rev 11371) @@ -23,11 +23,13 @@ public interface ConnectionLifeCycleListener { /** - * called when a connection is created. + * This method is used both by client connector creation and server connection creation through acceptors. + * the acceptor will be set to null on client operations * + * @param The acceptor here will be always null on a client connection created event. * @param connection the connection that has been created */ - void connectionCreated(Connection connection, ProtocolType protocol); + void connectionCreated(Acceptor acceptor, Connection connection, ProtocolType protocol); /** * called when a connection is destroyed. Modified: branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/integration/cluster/distribution/ClusterTestBase.java =================================================================== --- branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/integration/cluster/distribution/ClusterTestBase.java 2011-09-19 23:47:54 UTC (rev 11370) +++ branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/integration/cluster/distribution/ClusterTestBase.java 2011-09-20 00:50:32 UTC (rev 11371) @@ -940,10 +940,10 @@ for (ClusterConnection cc : clusterManager.getClusterConnections()) { out += cc.describe() + "\n"; + out += cc.getTopology().describe(); } } out += "\n\nfull topology:"; - out += clusterManager.getTopology().describe(); return out + br; } Modified: branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/integration/cluster/distribution/OnewayTwoNodeClusterTest.java =================================================================== --- branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/integration/cluster/distribution/OnewayTwoNodeClusterTest.java 2011-09-19 23:47:54 UTC (rev 11370) +++ branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/integration/cluster/distribution/OnewayTwoNodeClusterTest.java 2011-09-20 00:50:32 UTC (rev 11371) @@ -131,10 +131,6 @@ waitForTopology(servers[0], 2); waitForTopology(servers[1], 2); - - System.out.println(servers[0].getClusterManager().getTopology().describe()); - - System.out.println(servers[1].getClusterManager().getTopology().describe()); setupSessionFactory(0, isNetty(), true); setupSessionFactory(1, isNetty(), true); Modified: branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/integration/cluster/distribution/SimpleSymmetricClusterTest.java =================================================================== --- branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/integration/cluster/distribution/SimpleSymmetricClusterTest.java 2011-09-19 23:47:54 UTC (rev 11370) +++ branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/integration/cluster/distribution/SimpleSymmetricClusterTest.java 2011-09-20 00:50:32 UTC (rev 11371) @@ -153,17 +153,6 @@ waitForTopology(servers[1], 3); waitForTopology(servers[2], 3); - for (int i = 0 ; i < 3; i++) - { - System.out.println("top[" + i + "]=" + servers[i].getClusterManager().getTopology().describe()); - } - - for (int i = 0; i <= 2; i++) - { - log.info("*************************************\n " + servers[i] + - " topology:\n" + - servers[i].getClusterManager().getTopology().describe()); - } setupSessionFactory(0, isNetty()); setupSessionFactory(1, isNetty()); setupSessionFactory(2, isNetty()); @@ -196,12 +185,6 @@ startServers(0, 1); - for (int i = 0; i <= 1; i++) - { - log.info("*************************************\n " + servers[i] + - " topology:\n" + - servers[i].getClusterManager().getTopology().describe()); - } setupSessionFactory(0, isNetty()); setupSessionFactory(1, isNetty()); @@ -266,13 +249,6 @@ for (int i = 0; i <= 4; i++) { - log.info("*************************************\n " + servers[i] + - " topology:\n" + - servers[i].getClusterManager().getTopology().describe()); - } - - for (int i = 0; i <= 4; i++) - { setupSessionFactory(i, isNetty()); } Modified: branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/integration/cluster/distribution/TwoWayTwoNodeClusterTest.java =================================================================== --- branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/integration/cluster/distribution/TwoWayTwoNodeClusterTest.java 2011-09-19 23:47:54 UTC (rev 11370) +++ branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/integration/cluster/distribution/TwoWayTwoNodeClusterTest.java 2011-09-20 00:50:32 UTC (rev 11371) @@ -137,8 +137,6 @@ startServers(0, 1); waitForTopology(servers[0], 2); - System.out.println(servers[0].getClusterManager().getTopology().describe()); - System.out.println(servers[1].getClusterManager().getTopology().describe()); waitForTopology(servers[1], 2); for (int i = 0; i < 10; i++) @@ -148,7 +146,6 @@ log.info("#stop #test #" + i); stopServers(1); - System.out.println(servers[0].getClusterManager().getTopology().describe()); waitForTopology(servers[0], 1, 2000); log.info("#start #test #" + i); startServers(1); Modified: branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/integration/cluster/failover/FailBackAutoTest.java =================================================================== --- branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/integration/cluster/failover/FailBackAutoTest.java 2011-09-19 23:47:54 UTC (rev 11370) +++ branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/integration/cluster/failover/FailBackAutoTest.java 2011-09-20 00:50:32 UTC (rev 11371) @@ -28,6 +28,7 @@ import org.hornetq.api.core.client.ClientProducer; import org.hornetq.api.core.client.ClientSession; import org.hornetq.api.core.client.ClientSessionFactory; +import org.hornetq.api.core.client.ServerLocator; import org.hornetq.api.core.client.SessionFailureListener; import org.hornetq.core.client.impl.ClientSessionFactoryInternal; import org.hornetq.core.client.impl.ServerLocatorInternal; @@ -76,18 +77,28 @@ locator.setBlockOnDurableSend(true); locator.setFailoverOnInitialConnection(true); locator.setReconnectAttempts(-1); + ((ServerLocatorInternal)locator).setIdentity("testAutoFailback"); + ClientSessionFactoryInternal sf = createSessionFactoryAndWaitForTopology(locator, 2); final CountDownLatch latch = new CountDownLatch(1); ClientSession session = sendAndConsume(sf, true); + + System.out.println(locator.getTopology().describe()); MyListener listener = new MyListener(latch); session.addFailureListener(listener); + + System.out.println(locator.getTopology().describe()); liveServer.crash(); - + assertTrue(latch.await(5, TimeUnit.SECONDS)); + + log.info("backup (nowLive) topology = " + backupServer.getServer().getClusterManager().getDefaultConnection().getTopology().describe()); + + log.info("Server Crash!!!"); ClientProducer producer = session.createProducer(FailoverTestBase.ADDRESS); @@ -97,6 +108,11 @@ producer.send(message); + verifyMessageOnServer(1, 1); + + System.out.println(locator.getTopology().describe()); + + session.removeFailureListener(listener); final CountDownLatch latch2 = new CountDownLatch(1); @@ -107,6 +123,10 @@ log.info("******* starting live server back"); liveServer.start(); + + Thread.sleep(1000); + + System.out.println("After failback: " + locator.getTopology().describe()); assertTrue(latch2.await(5, TimeUnit.SECONDS)); @@ -118,6 +138,8 @@ session.close(); + verifyMessageOnServer(0, 1); + sf.close(); Assert.assertEquals(0, sf.numSessions()); @@ -125,6 +147,29 @@ Assert.assertEquals(0, sf.numConnections()); } + /** + * @throws Exception + * @throws HornetQException + */ + private void verifyMessageOnServer(final int server, final int numberOfMessages) throws Exception, HornetQException + { + ServerLocator backupLocator = createInVMLocator(server); + ClientSessionFactory factorybkp = backupLocator.createSessionFactory(); + ClientSession sessionbkp = factorybkp.createSession(false, false); + sessionbkp.start(); + ClientConsumer consumerbkp = sessionbkp.createConsumer(ADDRESS); + for (int i = 0 ; i < numberOfMessages; i++) + { + ClientMessage msg = consumerbkp.receive(1000); + assertNotNull(msg); + msg.acknowledge(); + sessionbkp.commit(); + } + sessionbkp.close(); + factorybkp.close(); + backupLocator.close(); + } + public void testAutoFailbackThenFailover() throws Exception { locator.setBlockOnNonDurableSend(true); @@ -253,7 +298,7 @@ if (createQueue) { - session.createQueue(FailoverTestBase.ADDRESS, FailoverTestBase.ADDRESS, null, false); + session.createQueue(FailoverTestBase.ADDRESS, FailoverTestBase.ADDRESS, null, true); } ClientProducer producer = session.createProducer(FailoverTestBase.ADDRESS); @@ -288,6 +333,8 @@ } ClientMessage message3 = consumer.receiveImmediate(); + + consumer.close(); Assert.assertNull(message3); @@ -315,6 +362,7 @@ public void connectionFailed(final HornetQException me, boolean failedOver) { + System.out.println("Failed, me"); latch.countDown(); } Modified: branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/integration/cluster/failover/GroupingFailoverTestBase.java =================================================================== --- branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/integration/cluster/failover/GroupingFailoverTestBase.java 2011-09-19 23:47:54 UTC (rev 11370) +++ branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/integration/cluster/failover/GroupingFailoverTestBase.java 2011-09-20 00:50:32 UTC (rev 11371) @@ -12,16 +12,9 @@ */ package org.hornetq.tests.integration.cluster.failover; -import java.util.concurrent.CountDownLatch; -import java.util.concurrent.TimeUnit; - -import junit.framework.Assert; - -import org.hornetq.api.core.HornetQException; import org.hornetq.api.core.Message; import org.hornetq.api.core.SimpleString; import org.hornetq.core.server.group.impl.GroupingHandlerConfiguration; -import org.hornetq.spi.core.protocol.RemotingConnection; import org.hornetq.tests.integration.cluster.distribution.ClusterTestBase; /** @@ -167,6 +160,7 @@ closeSessionFactory(0); Thread.sleep(1000); + servers[0].stop(true); waitForServerRestart(2); @@ -213,16 +207,4 @@ abstract boolean isSharedServer(); - private void fail(final RemotingConnection conn, final CountDownLatch latch) throws InterruptedException - { - // Simulate failure on connection - conn.fail(new HornetQException(HornetQException.NOT_CONNECTED)); - - // Wait to be informed of failure - - boolean ok = latch.await(1000, TimeUnit.MILLISECONDS); - - Assert.assertTrue(ok); - } - } Modified: branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/integration/cluster/failover/MultipleBackupsFailoverTestBase.java =================================================================== --- branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/integration/cluster/failover/MultipleBackupsFailoverTestBase.java 2011-09-19 23:47:54 UTC (rev 11370) +++ branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/integration/cluster/failover/MultipleBackupsFailoverTestBase.java 2011-09-20 00:50:32 UTC (rev 11371) @@ -193,7 +193,7 @@ { if (server != null) { - log.info("failed topology, Topology on server = " + server.getClusterManager().getTopology().describe()); + log.info("failed topology, Topology on server = " + server.getClusterManager().describe()); } } assertTrue("expected " + topologyMembers + " members", ok); Modified: branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/integration/cluster/failover/MultipleLivesMultipleBackupsFailoverTest.java =================================================================== --- branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/integration/cluster/failover/MultipleLivesMultipleBackupsFailoverTest.java 2011-09-19 23:47:54 UTC (rev 11370) +++ branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/integration/cluster/failover/MultipleLivesMultipleBackupsFailoverTest.java 2011-09-20 00:50:32 UTC (rev 11371) @@ -98,8 +98,6 @@ Thread.sleep(500); servers.get(0).crash(session); - System.out.println("server3 " + servers.get(3).getServer().getClusterManager().getTopology().describe()); - int liveAfter0 = waitForNewLive(10000, true, servers, 1, 2); ServerLocator locator2 = getServerLocator(3); Modified: branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/integration/http/NettyHttpTest.java =================================================================== --- branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/integration/http/NettyHttpTest.java 2011-09-19 23:47:54 UTC (rev 11370) +++ branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/integration/http/NettyHttpTest.java 2011-09-20 00:50:32 UTC (rev 11371) @@ -29,6 +29,7 @@ import org.hornetq.core.remoting.impl.netty.NettyConnector; import org.hornetq.core.remoting.impl.netty.TransportConstants; import org.hornetq.spi.core.protocol.ProtocolType; +import org.hornetq.spi.core.remoting.Acceptor; import org.hornetq.spi.core.remoting.BufferHandler; import org.hornetq.spi.core.remoting.Connection; import org.hornetq.spi.core.remoting.ConnectionLifeCycleListener; @@ -544,7 +545,7 @@ latch = connCreatedLatch; } - public void connectionCreated(final Connection connection, final ProtocolType protocol) + public void connectionCreated(final Acceptor acceptor, final Connection connection, final ProtocolType protocol) { this.connection = connection; if (latch != null) Modified: branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/integration/largemessage/mock/MockConnector.java =================================================================== --- branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/integration/largemessage/mock/MockConnector.java 2011-09-19 23:47:54 UTC (rev 11370) +++ branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/integration/largemessage/mock/MockConnector.java 2011-09-20 00:50:32 UTC (rev 11371) @@ -89,7 +89,7 @@ */ public MockConnection(final int serverID, final BufferHandler handler, final ConnectionLifeCycleListener listener) { - super(serverID, handler, listener, Executors.newSingleThreadExecutor()); + super(null, serverID, handler, listener, Executors.newSingleThreadExecutor()); } @Override Modified: branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/unit/core/remoting/impl/netty/NettyAcceptorFactoryTest.java =================================================================== --- branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/unit/core/remoting/impl/netty/NettyAcceptorFactoryTest.java 2011-09-19 23:47:54 UTC (rev 11370) +++ branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/unit/core/remoting/impl/netty/NettyAcceptorFactoryTest.java 2011-09-20 00:50:32 UTC (rev 11371) @@ -63,7 +63,7 @@ { } - public void connectionCreated(final Connection connection, final ProtocolType protocol) + public void connectionCreated(final Acceptor acceptor, final Connection connection, final ProtocolType protocol) { } @@ -74,7 +74,8 @@ }; - Acceptor acceptor = factory.createAcceptor(params, + Acceptor acceptor = factory.createAcceptor(null, + params, handler, null, listener, Modified: branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/unit/core/remoting/impl/netty/NettyAcceptorTest.java =================================================================== --- branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/unit/core/remoting/impl/netty/NettyAcceptorTest.java 2011-09-19 23:47:54 UTC (rev 11370) +++ branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/unit/core/remoting/impl/netty/NettyAcceptorTest.java 2011-09-20 00:50:32 UTC (rev 11371) @@ -28,6 +28,7 @@ import org.hornetq.core.remoting.impl.netty.NettyAcceptor; import org.hornetq.core.remoting.impl.netty.TransportConstants; import org.hornetq.spi.core.protocol.ProtocolType; +import org.hornetq.spi.core.remoting.Acceptor; import org.hornetq.spi.core.remoting.BufferHandler; import org.hornetq.spi.core.remoting.Connection; import org.hornetq.spi.core.remoting.ConnectionLifeCycleListener; @@ -80,7 +81,7 @@ { } - public void connectionCreated(final Connection connection, final ProtocolType protocol) + public void connectionCreated(final Acceptor acceptor, final Connection connection, final ProtocolType protocol) { } Modified: branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/unit/core/remoting/impl/netty/NettyConnectionTest.java =================================================================== --- branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/unit/core/remoting/impl/netty/NettyConnectionTest.java 2011-09-19 23:47:54 UTC (rev 11370) +++ branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/unit/core/remoting/impl/netty/NettyConnectionTest.java 2011-09-20 00:50:32 UTC (rev 11371) @@ -24,6 +24,7 @@ import org.hornetq.api.core.HornetQException; import org.hornetq.core.remoting.impl.netty.NettyConnection; import org.hornetq.spi.core.protocol.ProtocolType; +import org.hornetq.spi.core.remoting.Acceptor; import org.hornetq.spi.core.remoting.Connection; import org.hornetq.spi.core.remoting.ConnectionLifeCycleListener; import org.hornetq.tests.util.RandomUtil; @@ -220,7 +221,7 @@ class MyListener implements ConnectionLifeCycleListener { - public void connectionCreated(final Connection connection, final ProtocolType protocol) + public void connectionCreated(final Acceptor acceptor, final Connection connection, final ProtocolType protocol) { } Modified: branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/unit/core/remoting/impl/netty/NettyConnectorTest.java =================================================================== --- branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/unit/core/remoting/impl/netty/NettyConnectorTest.java 2011-09-19 23:47:54 UTC (rev 11370) +++ branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/unit/core/remoting/impl/netty/NettyConnectorTest.java 2011-09-20 00:50:32 UTC (rev 11371) @@ -23,6 +23,7 @@ import org.hornetq.api.core.HornetQException; import org.hornetq.core.remoting.impl.netty.NettyConnector; import org.hornetq.spi.core.protocol.ProtocolType; +import org.hornetq.spi.core.remoting.Acceptor; import org.hornetq.spi.core.remoting.BufferHandler; import org.hornetq.spi.core.remoting.Connection; import org.hornetq.spi.core.remoting.ConnectionLifeCycleListener; @@ -66,7 +67,7 @@ { } - public void connectionCreated(final Connection connection, final ProtocolType protocol) + public void connectionCreated(final Acceptor acceptor, final Connection connection, final ProtocolType protocol) { } public void connectionReadyForWrites(Object connectionID, boolean ready) @@ -106,7 +107,7 @@ { } - public void connectionCreated(final Connection connection, final ProtocolType protocol) + public void connectionCreated(final Acceptor acceptor, final Connection connection, final ProtocolType protocol) { } Modified: branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/util/ServiceTestBase.java =================================================================== --- branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/util/ServiceTestBase.java 2011-09-19 23:47:54 UTC (rev 11370) +++ branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/util/ServiceTestBase.java 2011-09-20 00:50:32 UTC (rev 11371) @@ -19,6 +19,7 @@ import java.util.HashMap; import java.util.List; import java.util.Map; +import java.util.Set; import javax.management.MBeanServer; @@ -37,11 +38,13 @@ import org.hornetq.core.remoting.impl.invm.InVMAcceptorFactory; import org.hornetq.core.remoting.impl.invm.InVMConnectorFactory; import org.hornetq.core.remoting.impl.invm.InVMRegistry; +import org.hornetq.core.remoting.impl.invm.TransportConstants; import org.hornetq.core.remoting.impl.netty.NettyAcceptorFactory; import org.hornetq.core.remoting.impl.netty.NettyConnectorFactory; import org.hornetq.core.server.HornetQServer; import org.hornetq.core.server.HornetQServers; import org.hornetq.core.server.NodeManager; +import org.hornetq.core.server.cluster.ClusterConnection; import org.hornetq.core.server.impl.HornetQServerImpl; import org.hornetq.core.settings.impl.AddressFullMessagePolicy; import org.hornetq.core.settings.impl.AddressSettings; @@ -112,8 +115,15 @@ log.debug("waiting for " + nodes + " on the topology for server = " + server); long start = System.currentTimeMillis(); + + Set<ClusterConnection> ccs = server.getClusterManager().getClusterConnections(); + + if (ccs.size() != 1) + { + throw new IllegalStateException("You need a single cluster connection on this version of waitForTopology on ServiceTestBase"); + } - Topology topology = server.getClusterManager().getTopology(); + Topology topology = ccs.iterator().next().getTopology(); do { @@ -521,7 +531,19 @@ locators.add(locatorWithoutHA); return locatorWithoutHA; } + + protected ServerLocator createInVMLocator(final int serverID) + { + Map<String, Object> server1Params = new HashMap<String, Object>(); + if (serverID != 0) + { + server1Params.put(TransportConstants.SERVER_ID_PROP_NAME, serverID); + } + + return HornetQClient.createServerLocatorWithHA(new TransportConfiguration(INVM_CONNECTOR_FACTORY, server1Params)); + } + protected ClientSessionFactoryImpl createFactory(final String connectorClass) throws Exception { ServerLocator locator = HornetQClient.createServerLocatorWithoutHA(new TransportConfiguration(connectorClass));