tweaks on tests

hornetq · Aug 14, 2011 · f488687 · f488687
1 parent 05f7adc
commit f488687
Show file tree

Hide file tree

Showing 3 changed files with 27 additions and 81 deletions.
diff --git a/src/main/org/hornetq/core/client/impl/ClientSessionFactoryImpl.java b/src/main/org/hornetq/core/client/impl/ClientSessionFactoryImpl.java
@@ -662,7 +662,10 @@ private void failoverOrReconnect(final Object connectionID, final HornetQExcepti
 
             reconnectSessions(oldConnection, reconnectAttempts);
 
-            oldConnection.destroy();
+            if (oldConnection != null)
+            {
+               oldConnection.destroy();
+            }
          }
          else
          {
@@ -1342,16 +1345,28 @@ public ConnectorFactory run()
 
    private void lockChannel1()
    {
-      Channel channel1 = connection.getChannel(1, -1);
-
-      channel1.getLock().lock();
+      if (connection != null)
+      {
+         Channel channel1 = connection.getChannel(1, -1);
+
+         if (channel1 != null)
+         {
+            channel1.getLock().lock();
+         }
+      }
    }
 
    private void unlockChannel1()
    {
-      Channel channel1 = connection.getChannel(1, -1);
-
-      channel1.getLock().unlock();
+      if (connection != null)
+      {
+         Channel channel1 = connection.getChannel(1, -1);
+
+         if (channel1 != null)
+         {
+            channel1.getLock().unlock();
+         }
+      }
    }
 
    private void forceReturnChannel1()

diff --git a/src/main/org/hornetq/core/server/cluster/impl/ClusterConnectionImpl.java b/src/main/org/hornetq/core/server/cluster/impl/ClusterConnectionImpl.java
@@ -658,9 +658,7 @@ private void createNewRecord(final String targetNodeID,
                                 final Queue queue,
                                 final boolean start) throws Exception
    {
-      final Topology topology = new Topology(null);
-      topology.setExecutor(executorFactory.getExecutor());
-      final ServerLocatorInternal targetLocator = new ServerLocatorImpl(topology, false, connector);
+      final ServerLocatorInternal targetLocator = new ServerLocatorImpl(clusterManagerTopology, false, connector);
 
       targetLocator.setReconnectAttempts(0);
 
@@ -687,54 +685,8 @@ private void createNewRecord(final String targetNodeID,
       }
 
       targetLocator.disableFinalizeCheck();
-
-      final ClusterTopologyListener listenerOnBridgeTopology = new ClusterTopologyListener()
-      {
-
-         public void nodeDown(String nodeID)
-         {
-            clusterManagerTopology.removeMember(nodeID);
-         }
-
-         public void nodeUP(String nodeID,
-                            Pair<TransportConfiguration, TransportConfiguration> connectorPair,
-                            boolean last)
-         {
-            clusterManagerTopology.addMember(nodeID, new TopologyMember(connectorPair), last);
-         }
-
-      };
-
-      final ClusterTopologyListener listenerOnMainTopology = new ClusterTopologyListener()
-      {
-         public void nodeDown(String nodeID)
-         {
-            topology.removeMember(nodeID);
-         }
-
-         public void nodeUP(String nodeID,
-                            Pair<TransportConfiguration, TransportConfiguration> connectorPair,
-                            boolean last)
-         {
-            topology.addMember(nodeID, new TopologyMember(connectorPair), last);
-         }
-
-      };
-
-      // Establish a proxy between each other topology
-      topology.addClusterTopologyListener(listenerOnBridgeTopology);
-
-      clusterManagerTopology.addClusterTopologyListener(listenerOnMainTopology);
-
-      MessageFlowRecordImpl record = new MessageFlowRecordImpl(listenerOnMainTopology,
-                                                               listenerOnBridgeTopology,
-                                                               targetLocator,
-                                                               targetNodeID,
-                                                               connector,
-                                                               queueName,
-                                                               queue);
-
-      topology.setOwner(record);
+
+      MessageFlowRecordImpl record = new MessageFlowRecordImpl(targetLocator, targetNodeID, connector, queueName, queue);
 
       ClusterConnectionBridge bridge = new ClusterConnectionBridge(this,
                                                                    manager,
@@ -777,14 +729,6 @@ public void nodeUP(String nodeID,
       if (start)
       {
          bridge.start();
-
-         bridge.getExecutor().execute(new Runnable(){
-            public void run()
-            {
-               topology.sendTopology(listenerOnBridgeTopology);
-               clusterManagerTopology.sendTopology(listenerOnMainTopology);
-            }
-         });
       }
    }
 
@@ -804,8 +748,6 @@ private class MessageFlowRecordImpl implements MessageFlowRecord
 
       private boolean disconnected = false;
 
-      private boolean sentInitialTopology = false;
-
       private final Queue queue;
 
       private final Map<SimpleString, RemoteQueueBinding> bindings = new HashMap<SimpleString, RemoteQueueBinding>();
@@ -814,13 +756,7 @@ private class MessageFlowRecordImpl implements MessageFlowRecord
 
       private volatile boolean firstReset = false;
 
-      private final ClusterTopologyListener listenerOnMainTopology;
-
-      private final ClusterTopologyListener listenerOnBridgeTopology;
-
-      public MessageFlowRecordImpl(final ClusterTopologyListener listenerOnMainTopology,
-                                   final ClusterTopologyListener listenerOnBridgeTopology,
-                                   final ServerLocatorInternal targetLocator,
+      public MessageFlowRecordImpl(final ServerLocatorInternal targetLocator,
                                    final String targetNodeID,
                                    final TransportConfiguration connector,
                                    final SimpleString queueName,
@@ -831,8 +767,6 @@ public MessageFlowRecordImpl(final ClusterTopologyListener listenerOnMainTopolog
          this.targetNodeID = targetNodeID;
          this.connector = connector;
          this.queueName = queueName;
-         this.listenerOnMainTopology = listenerOnMainTopology;
-         this.listenerOnBridgeTopology = listenerOnBridgeTopology;
       }
 
       /* (non-Javadoc)
@@ -909,9 +843,6 @@ public void close() throws Exception
             log.trace("Stopping bridge " + bridge);
          }
 
-         clusterManagerTopology.removeClusterTopologyListener(listenerOnMainTopology);
-         targetLocator.getTopology().removeClusterTopologyListener(listenerOnBridgeTopology);
-
          isClosed = true;
          clearBindings();
 

diff --git a/tests/src/org/hornetq/tests/integration/cluster/distribution/ClusterTestBase.java b/tests/src/org/hornetq/tests/integration/cluster/distribution/ClusterTestBase.java
@@ -2079,7 +2079,7 @@ protected void stopServers(final int... nodes) throws Exception
             {
                ClusterTestBase.log.info("stopping server " + node);
                servers[node].stop();
-               Thread.sleep(1000);
+               Thread.sleep(500);
                ClusterTestBase.log.info("server " + node + " stopped");
             }
             catch (Exception e)