[hornetq-commits] JBoss hornetq SVN: r8525 - in trunk/src/main/org/hornetq/core: remoting/impl and 3 other directories.

[do-not-reply at jboss.org]

Author: jmesnil
Date: 2009-12-03 10:07:58 -0500 (Thu, 03 Dec 2009)
New Revision: 8525

Modified:
   trunk/src/main/org/hornetq/core/remoting/RemotingConnection.java
   trunk/src/main/org/hornetq/core/remoting/impl/RemotingConnectionImpl.java
   trunk/src/main/org/hornetq/core/remoting/server/RemotingService.java
   trunk/src/main/org/hornetq/core/remoting/server/impl/RemotingServiceImpl.java
   trunk/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java
Log:
server shutdown

* reverted r8522
* make sure that after acceptors are stopped, all the channels are cleared
  before sending DISCONNECT and closing the connections

Modified: trunk/src/main/org/hornetq/core/remoting/RemotingConnection.java
===================================================================
--- trunk/src/main/org/hornetq/core/remoting/RemotingConnection.java	2009-12-03 14:05:00 UTC (rev 8524)
+++ trunk/src/main/org/hornetq/core/remoting/RemotingConnection.java	2009-12-03 15:07:58 UTC (rev 8525)
@@ -73,4 +73,6 @@
    Object getTransferLock(); 
    
    boolean checkDataReceived();
+
+   void removeAllChannels();
 }

Modified: trunk/src/main/org/hornetq/core/remoting/impl/RemotingConnectionImpl.java
===================================================================
--- trunk/src/main/org/hornetq/core/remoting/impl/RemotingConnectionImpl.java	2009-12-03 14:05:00 UTC (rev 8524)
+++ trunk/src/main/org/hornetq/core/remoting/impl/RemotingConnectionImpl.java	2009-12-03 15:07:58 UTC (rev 8525)
@@ -322,6 +322,16 @@
 
       return res;
    }
+   
+   public void removeAllChannels()
+   {
+      //We get the transfer lock first - this ensures no packets are being processed AND
+      //it's guaranteed no more packets will be processed once this method is complete
+      synchronized (transferLock)
+      {
+         channels.clear();
+      }
+   }
 
    // Buffer Handler implementation
    // ----------------------------------------------------
@@ -451,5 +461,4 @@
          channel.close();
       }
    }
-
 }

Modified: trunk/src/main/org/hornetq/core/remoting/server/RemotingService.java
===================================================================
--- trunk/src/main/org/hornetq/core/remoting/server/RemotingService.java	2009-12-03 14:05:00 UTC (rev 8524)
+++ trunk/src/main/org/hornetq/core/remoting/server/RemotingService.java	2009-12-03 15:07:58 UTC (rev 8525)
@@ -45,6 +45,4 @@
    void freeze();
 
    RemotingConnection getServerSideReplicatingConnection();
-
-   void pause() throws Exception;
 }

Modified: trunk/src/main/org/hornetq/core/remoting/server/impl/RemotingServiceImpl.java
===================================================================
--- trunk/src/main/org/hornetq/core/remoting/server/impl/RemotingServiceImpl.java	2009-12-03 14:05:00 UTC (rev 8524)
+++ trunk/src/main/org/hornetq/core/remoting/server/impl/RemotingServiceImpl.java	2009-12-03 15:07:58 UTC (rev 8525)
@@ -209,13 +209,18 @@
          }
       }
    }
-
-   public void pause() throws Exception
+   
+   public void stop() throws Exception
    {
       if (!started)
       {
          return;
       }
+      
+      if (!started)
+      {
+         return;
+      }
 
       failureCheckThread.close();
 
@@ -224,40 +229,41 @@
       {
          acceptor.pause();
       }
-   }
-   
-   public void stop() throws Exception
-   {
-      if (!started)
-      {
-         return;
-      }
 
-      synchronized (server)
+      //Now we ensure that no connections will process any more packets after this method is complete
+      //then send a disconnect packet
+      for (ConnectionEntry entry : connections.values())
       {
-         for (ConnectionEntry entry : connections.values())
-         {
-            entry.connection.getChannel(0, -1).sendAndFlush(new PacketImpl(DISCONNECT));
-         }
+         RemotingConnection conn = entry.connection;
 
-         for (Acceptor acceptor : acceptors)
-         {
-            acceptor.stop();
-         }
+         Channel channel0 = conn.getChannel(0, -1);
 
-         acceptors.clear();
+         //And we remove all channels from the connection, this ensures no more packets will be processed after this method is
+         //complete
 
-         connections.clear();
+         conn.removeAllChannels();
 
-         if (managementService != null)
-         {
-            managementService.unregisterAcceptors();
-         }
+         //Now we are 100% sure that no more packets will be processed we can send the disconnect
 
-         started = false;
+         channel0.sendAndFlush(new PacketImpl(DISCONNECT));
+      }
+      
+      for (Acceptor acceptor : acceptors)
+      {
+         acceptor.stop();
+      }
 
+      acceptors.clear();
+
+      connections.clear();
+
+      if (managementService != null)
+      {
+         managementService.unregisterAcceptors();
       }
 
+      started = false;
+
    }
 
    public boolean isStarted()

Modified: trunk/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java
===================================================================
--- trunk/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java	2009-12-03 14:05:00 UTC (rev 8524)
+++ trunk/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java	2009-12-03 15:07:58 UTC (rev 8525)
@@ -357,14 +357,6 @@
       }
 
       // we stop the remoting service outside a lock
-      remotingService.pause();
-
-      if (replicationManager != null)
-      {
-         replicationManager.stop();
-         replicationManager = null;
-      }
-      
       remotingService.stop();
 
       synchronized (this)
@@ -402,6 +394,12 @@
          {
             storageManager.stop();
          }
+         
+         if (replicationManager != null)
+         {
+            replicationManager.stop();
+            replicationManager = null;
+         }
 
          if (replicationEndpoint != null)
          {