[jboss-cvs] JBoss Messaging SVN: r3248 - in trunk/src/main/org/jboss/jms: client/container and 2 other directories.

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

Author: timfox
Date: 2007-10-26 16:14:06 -0400 (Fri, 26 Oct 2007)
New Revision: 3248

Modified:
   trunk/src/main/org/jboss/jms/client/FailoverCommandCenter.java
   trunk/src/main/org/jboss/jms/client/container/ConsumerAspect.java
   trunk/src/main/org/jboss/jms/client/container/FailoverValveInterceptor.java
   trunk/src/main/org/jboss/jms/client/delegate/ClientConnectionDelegate.java
   trunk/src/main/org/jboss/jms/delegate/ConnectionDelegate.java
Log:
Fixed issue with MergeQueueTest


Modified: trunk/src/main/org/jboss/jms/client/FailoverCommandCenter.java
===================================================================
--- trunk/src/main/org/jboss/jms/client/FailoverCommandCenter.java	2007-10-26 17:52:01 UTC (rev 3247)
+++ trunk/src/main/org/jboss/jms/client/FailoverCommandCenter.java	2007-10-26 20:14:06 UTC (rev 3248)
@@ -144,22 +144,27 @@
             log.trace("Synchronizing state");
             state.getDelegate().synchronizeWith(newDelegate);
             log.trace("Synchronized state");
-                           
-            log.trace("Opening valve");
-            valve.open();
-            log.trace("Opened valve");
-            valveOpened = true;
             
-            //Now start the connection - note! this can't be done while the valve is closed
-            //or it will block itself
+            //Now restart the connection if appropriate
+            //Note! we mus start the connection while the valve is still closed
+            //Otherwise If a consumer closing is waiting on failover to complete
+            //Then on failover complete the valve will be opened and closing retried on a
+            //different thread
+            //but the next line will re-start the connection so there is a race between the two
+            //If the restart hits after closing then messages can get delivered after consumer
+            //is closed
             
-            // start the connection again on the serverEndpoint if necessary            
             if (state.isStarted())
             {
-            	log.trace("Starting new connection");
-               newDelegate.start();
+               log.trace("Starting new connection");
+               newDelegate.startAfterFailover();
                log.trace("Started new connection");
             }
+                           
+            log.trace("Opening valve");
+            valve.open();
+            log.trace("Opened valve");
+            valveOpened = true;
             
             failoverSuccessful = true;      
             

Modified: trunk/src/main/org/jboss/jms/client/container/ConsumerAspect.java
===================================================================
--- trunk/src/main/org/jboss/jms/client/container/ConsumerAspect.java	2007-10-26 17:52:01 UTC (rev 3247)
+++ trunk/src/main/org/jboss/jms/client/container/ConsumerAspect.java	2007-10-26 20:14:06 UTC (rev 3248)
@@ -33,6 +33,7 @@
 import org.jboss.jms.delegate.ConnectionDelegate;
 import org.jboss.jms.delegate.ConsumerDelegate;
 import org.jboss.jms.delegate.SessionDelegate;
+import org.jboss.logging.Logger;
 import org.jboss.messaging.util.MessageQueueNameHelper;
 
 import EDU.oswego.cs.dl.util.concurrent.QueuedExecutor;
@@ -53,6 +54,9 @@
 {
    // Constants ------------------------------------------------------------------------------------
    
+   private static final Logger log = Logger.getLogger(ConsumerAspect.class);
+
+   
    // Static ---------------------------------------------------------------------------------------
 
    // Attributes -----------------------------------------------------------------------------------

Modified: trunk/src/main/org/jboss/jms/client/container/FailoverValveInterceptor.java
===================================================================
--- trunk/src/main/org/jboss/jms/client/container/FailoverValveInterceptor.java	2007-10-26 17:52:01 UTC (rev 3247)
+++ trunk/src/main/org/jboss/jms/client/container/FailoverValveInterceptor.java	2007-10-26 20:14:06 UTC (rev 3248)
@@ -96,6 +96,12 @@
       
       JMSRemotingConnection remotingConnection = null;
       String methodName = ((MethodInvocation)invocation).getMethod().getName();
+      
+      if (methodName.equals("startAfterFailover"))
+      {
+         //We don't use the valve on this method
+         return invocation.invokeNext();
+      }
 
       boolean left = false;
       

Modified: trunk/src/main/org/jboss/jms/client/delegate/ClientConnectionDelegate.java
===================================================================
--- trunk/src/main/org/jboss/jms/client/delegate/ClientConnectionDelegate.java	2007-10-26 17:52:01 UTC (rev 3247)
+++ trunk/src/main/org/jboss/jms/client/delegate/ClientConnectionDelegate.java	2007-10-26 20:14:06 UTC (rev 3248)
@@ -243,7 +243,14 @@
 
       doInvoke(client, req);
    }
+   
+   public void startAfterFailover() throws JMSException
+   {
+      RequestSupport req = new ConnectionStartRequest(id, version);
 
+      doInvoke(client, req);
+   }
+
    public void stop() throws JMSException
    {
       RequestSupport req = new ConnectionStopRequest(id, version);

Modified: trunk/src/main/org/jboss/jms/delegate/ConnectionDelegate.java
===================================================================
--- trunk/src/main/org/jboss/jms/delegate/ConnectionDelegate.java	2007-10-26 17:52:01 UTC (rev 3247)
+++ trunk/src/main/org/jboss/jms/delegate/ConnectionDelegate.java	2007-10-26 20:14:06 UTC (rev 3248)
@@ -57,4 +57,8 @@
    void registerFailoverListener(FailoverListener failoverListener);
    
    boolean unregisterFailoverListener(FailoverListener failoverListener);
+   
+   //Special start method that doesn't get intercepted to allow connection to be restarted
+   //after failover when the valve is still held
+   void startAfterFailover() throws JMSException;
 }