9:07 a.m.
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)
{