[jboss-cvs] JBoss Messaging SVN: r8185 - branches/JBM1842/src/main/org/jboss/messaging/core/impl/postoffice.

[jboss-cvs-commits at lists.jboss.org]

Author: gaohoward
Date: 2011-01-21 08:07:12 -0500 (Fri, 21 Jan 2011)
New Revision: 8185

Modified:
   branches/JBM1842/src/main/org/jboss/messaging/core/impl/postoffice/MessagingPostOffice.java
Log:
save


Modified: branches/JBM1842/src/main/org/jboss/messaging/core/impl/postoffice/MessagingPostOffice.java
===================================================================
--- branches/JBM1842/src/main/org/jboss/messaging/core/impl/postoffice/MessagingPostOffice.java	2011-01-21 11:22:01 UTC (rev 8184)
+++ branches/JBM1842/src/main/org/jboss/messaging/core/impl/postoffice/MessagingPostOffice.java	2011-01-21 13:07:12 UTC (rev 8185)
@@ -254,9 +254,6 @@
    private Object viewUpdateLock;
    private boolean stopUpdate = false;
    private boolean updateInProcess = false;
-      
-   //state
-//   private int state = STATE_DEAD;
    
    private StateMonitor stateMonitor = null;
    
@@ -441,12 +438,8 @@
 	      
 	      groupMember.multicastControl(new JoinClusterRequest(thisNodeID, info), true);
 	      
-	      initState(STATE_CLUSTERED);
+	      initClusterState();
       }
-      else
-      {
-         initState(STATE_STANDALONE);
-      }
    
       //Now load the bindings for this node
       
@@ -458,29 +451,24 @@
    }
    
    //this method will trigger a dedicated thread to write time stamp
-   private synchronized void initState(int newState) throws Exception
+   private synchronized void initClusterState() throws Exception
    {
       if (this.keepOldFailoverModel) return;
       
-      if (STATE_CLUSTERED == newState)
-      {
-         this.updateStateInStorage(thisNodeID, newState);
+      this.addThisNodeStateInStorage();
          
-         if (stateMonitor == null)
-         {
-            stateMonitor = new StateMonitor();
-            stateMonitor.start();
-         }
-      }
+      stateMonitor = new StateMonitor();
+      stateMonitor.start();
    }
    
-   private void updateStateInStorage(final int nID, final int newState) throws Exception
+   private void addThisNodeStateInStorage() throws Exception
    {
       if (ds == null)
       {
          return;
       }
-      class UpdateState extends JDBCTxRunner
+
+      class AddNodeState extends JDBCTxRunner
       {
          public Object doTransaction() throws Exception
          {
@@ -491,18 +479,18 @@
             {
                ps = conn.prepareStatement(getSQLStatement("UPDATE_STATE"));
 
-               ps.setInt(1, newState);
-               ps.setInt(2, nID);
+               ps.setInt(1, STATE_CLUSTERED);
+               ps.setInt(2, thisNodeID);
 
                int row = ps.executeUpdate();
                
                if (row == 0)
                {
                   ps1 = conn.prepareStatement(getSQLStatement("INSERT_NODE_STATE"));
-                  ps1.setInt(1, nID);
+                  ps1.setInt(1, thisNodeID);
                   ps1.setLong(2, System.currentTimeMillis());
-                  ps1.setInt(3, newState);
-                  
+                  ps1.setInt(3, STATE_CLUSTERED);
+
                   ps1.executeUpdate();
                }
             }
@@ -514,9 +502,42 @@
             return null;
          }
       }
-      new UpdateState().executeWithRetry();
+      new AddNodeState().executeWithRetry();
    }
    
+   private Integer updateStateInStorage(final int nID, final int newState) throws Exception
+   {
+      if (ds == null)
+      {
+         return 0;
+      }
+
+      class UpdateState extends JDBCTxRunner
+      {
+         public Object doTransaction() throws Exception
+         {
+            PreparedStatement ps  = null;
+            int row = 0;
+            
+            try
+            {
+               ps = conn.prepareStatement(getSQLStatement("UPDATE_STATE"));
+
+               ps.setInt(1, newState);
+               ps.setInt(2, nID);
+
+               row = ps.executeUpdate();
+            }
+            finally
+            {
+               closeStatement(ps);
+            }
+            return new Integer(row);
+         }
+      }
+      return (Integer)new UpdateState().executeWithRetry();
+   }
+   
    private void cleanUpSuspectedNode(final Integer qNode) throws Exception
    {
       QuarantinedNode node = suspectedNodes.remove(qNode);
@@ -573,47 +594,36 @@
          
          if (isDead)
          {
-            //update node status. this definitely not me!
-            this.updateStateInStorage(qNodeID, STATE_DEAD);
-            
+
             QuarantinedNode qNode = suspectedNodes.get(qNodeID);
             Integer fNodeID = qNode.getFailover();
-            
+
             Integer foNodeID = (Integer)failoverMap.get(fNodeID);
             if (foNodeID == null)
             {
                throw new IllegalStateException("Failover node " + fNodeID + " for node " + qNode + " is not alive!");
             }
-            
+
+            //every suspected node must have a one and only failover node.
             if (fNodeID.intValue() == thisNodeID)
             {
-               //I am the failover node for the dead, perform failover now
+               // I am the failover node for the dead, perform failover now
                if (suspectedNodes.get(qNodeID).shouldFailover() && isSupportsFailover())
                {
-                   log.debug(this + ": I am the failover node for node " + qNodeID + " that crashed");
-            
-                   performFailover(qNodeID);
-                   
-                   //now clean up the quarantined set
-                   cleanUpSuspectedNode(qNodeID);
-                   
-                   //broadcast
-                   NodeDeadRequest request = new NodeDeadRequest(qNodeID);
-                   
-                   groupMember.multicastControl(request, true);
-               }
-            }
-            else
-            {
-               //if all nodes becomes quarantined, including myself, I have to try claim the dead also
-               if (clusterState.allDeadButMe(thisNodeID))
-               {
-                  log.info(this + ": I am the only one alive, so pick up the dead ones");
+                  log.debug(this + ": I am the failover node for node " + qNodeID + " that crashed");
+
+                  //update node status. this definitely not me!
+                  this.updateStateInStorage(qNodeID, STATE_DEAD);
+
                   performFailover(qNodeID);
-                  
+
+                  // now clean up the quarantined set
                   cleanUpSuspectedNode(qNodeID);
-                  
-                  //no need to broadcast as no one is listening.
+
+                  // broadcast
+                  NodeDeadRequest request = new NodeDeadRequest(qNodeID);
+
+                  groupMember.multicastControl(request, true);
                }
             }
          }