[jboss-cvs] JBossAS SVN: r91228 - branches/JBPAPP_5_0/connector/src/main/org/jboss/resource/adapter/jms/inflow.

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

Author: jesper.pedersen
Date: 2009-07-14 11:31:43 -0400 (Tue, 14 Jul 2009)
New Revision: 91228

Modified:
   branches/JBPAPP_5_0/connector/src/main/org/jboss/resource/adapter/jms/inflow/JmsActivationSpec.java
   branches/JBPAPP_5_0/connector/src/main/org/jboss/resource/adapter/jms/inflow/JmsServerSession.java
Log:
[JBPAPP-2260] Setting maxMessages > 1 results in multiple message deliveries within one transaction

Modified: branches/JBPAPP_5_0/connector/src/main/org/jboss/resource/adapter/jms/inflow/JmsActivationSpec.java
===================================================================
--- branches/JBPAPP_5_0/connector/src/main/org/jboss/resource/adapter/jms/inflow/JmsActivationSpec.java	2009-07-14 15:26:56 UTC (rev 91227)
+++ branches/JBPAPP_5_0/connector/src/main/org/jboss/resource/adapter/jms/inflow/JmsActivationSpec.java	2009-07-14 15:31:43 UTC (rev 91228)
@@ -129,6 +129,8 @@
    
    private int forceClearAttempts = 0;
    
+   private Boolean forceTransacted = Boolean.FALSE;
+   
    public void setForceClearOnShutdown(boolean forceClear)
    {
       this.forceClearOnShutdown = forceClear;
@@ -715,4 +717,17 @@
    {
       this.isSameRMOverrideValue = isSameRMOverrideValue;
    }
+
+   public Boolean isForceTransacted()
+   {
+      if (forceTransacted != null)
+         return forceTransacted;
+
+      return Boolean.FALSE;
+   }
+
+   public void setForceTransacted(Boolean forceTransacted)
+   {
+      this.forceTransacted = forceTransacted;
+   }
 }

Modified: branches/JBPAPP_5_0/connector/src/main/org/jboss/resource/adapter/jms/inflow/JmsServerSession.java
===================================================================
--- branches/JBPAPP_5_0/connector/src/main/org/jboss/resource/adapter/jms/inflow/JmsServerSession.java	2009-07-14 15:26:56 UTC (rev 91227)
+++ branches/JBPAPP_5_0/connector/src/main/org/jboss/resource/adapter/jms/inflow/JmsServerSession.java	2009-07-14 15:31:43 UTC (rev 91228)
@@ -167,7 +167,8 @@
    {
       try
       {
-         endpoint.beforeDelivery(JmsActivation.ONMESSAGE);
+         if (!(txnStrategy != null && txnStrategy instanceof TraditionalXATransactionDemarcationStrategy))
+            endpoint.beforeDelivery(JmsActivation.ONMESSAGE);
 
          try
          {
@@ -179,7 +180,8 @@
             }
          } finally
          {
-            endpoint.afterDelivery();
+            if (!(txnStrategy != null && txnStrategy instanceof TraditionalXATransactionDemarcationStrategy))
+               endpoint.afterDelivery();
 
             if (dlqHandler != null)
                dlqHandler.messageDelivered(message);
@@ -277,41 +279,48 @@
 
    private class DemarcationStrategyFactory
    {
-
       TransactionDemarcationStrategy getStrategy()
       {
          TransactionDemarcationStrategy current = null;
-			final JmsActivationSpec spec = pool.getActivation()
-					.getActivationSpec();
-			final JmsActivation activation = pool.getActivation();
-			try 
-			{
-				//If we have a transacted delivery
-        	    if (activation.isDeliveryTransacted() )
-				{
-					//if we have an XASession
-        	    	if(xaSession != null)
-					{
-						current = new XATransactionDemarcationStrategy();
-					}
-					else  //if we don't have an XASession, simulate it with a transacted session
-					{
-						current = new SimulatedXATransactionDemarcationStrategy();
-					}
-					
-				}
-        	    else
-				{
-					current = new LocalDemarcationStrategy();
-				}
-			} 
-			catch (Throwable t) 
-			{
-				log.error(this + " error creating transaction demarcation ", t);
-			}
-			return current;
-		}
+         final JmsActivationSpec spec = pool.getActivation().getActivationSpec();
+         final JmsActivation activation = pool.getActivation();
+         try 
+         {
+            //If we have a transacted delivery
+            if (activation.isDeliveryTransacted())
+            {
+               //if we have an XASession
+               if (xaSession != null)
+               {
+                  if (spec.isForceTransacted())
+                  {
+                     current = new XATransactionDemarcationStrategy();
+                  }
+                  else
+                  {
+                     current = new TraditionalXATransactionDemarcationStrategy();
+                  }
+               }
+               else  //if we don't have an XASession, simulate it with a transacted session
+               {
+                  current = new SimulatedXATransactionDemarcationStrategy();
+               }
+            }
+            else
+            {
+               current = new LocalDemarcationStrategy();
+            }
+         } 
+         catch (Throwable t) 
+         {
+            log.error(this + " error creating transaction demarcation ", t);
+         }
 
+         if (current != null && log.isTraceEnabled())
+            log.trace("Using strategy: " + current.getClass().getName());
+
+         return current;
+      }
    }
 
    private interface TransactionDemarcationStrategy
@@ -686,4 +695,32 @@
 		}
 
 	}
+
+   /**
+    * This class is used for traditional XATransaction interaction as described in JCA 1.5 12.5.6
+    */
+   private class TraditionalXATransactionDemarcationStrategy implements TransactionDemarcationStrategy
+   {
+      boolean trace = log.isTraceEnabled();
+      TransactionManager tm = pool.getActivation().getTransactionManager();;
+
+      public TraditionalXATransactionDemarcationStrategy() throws Throwable
+      {
+         final int timeout = pool.getActivation().getActivationSpec().getTransactionTimeout();
+         
+         if (timeout > 0)
+         {
+            log.trace("Setting transactionTimeout for JMSSessionPool to " + timeout);
+            tm.setTransactionTimeout(timeout);
+         }
+      }
+
+      public void error()
+      {
+      }
+
+      public void end()
+      {
+      }
+   }
 }
\ No newline at end of file