[jboss-cvs] JBoss Messaging SVN: r5178 - branches/Branch_JBMESSAGING_1416/src/main/org/jboss/messaging/core/impl.

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

Author: gaohoward
Date: 2008-10-24 06:34:38 -0400 (Fri, 24 Oct 2008)
New Revision: 5178

Modified:
   branches/Branch_JBMESSAGING_1416/src/main/org/jboss/messaging/core/impl/ChannelSupport.java
   branches/Branch_JBMESSAGING_1416/src/main/org/jboss/messaging/core/impl/OrderingGroup.java
   branches/Branch_JBMESSAGING_1416/src/main/org/jboss/messaging/core/impl/OrderingGroupMonitor.java
Log:
JBMESSAGING-1416


Modified: branches/Branch_JBMESSAGING_1416/src/main/org/jboss/messaging/core/impl/ChannelSupport.java
===================================================================
--- branches/Branch_JBMESSAGING_1416/src/main/org/jboss/messaging/core/impl/ChannelSupport.java	2008-10-23 22:05:06 UTC (rev 5177)
+++ branches/Branch_JBMESSAGING_1416/src/main/org/jboss/messaging/core/impl/ChannelSupport.java	2008-10-24 10:34:38 UTC (rev 5178)
@@ -32,6 +32,8 @@
 import javax.jms.Message;
 import javax.jms.TextMessage;
 
+import com.arjuna.ats.jta.TransactionManager;
+
 import org.jboss.jms.message.JBossMessage;
 import org.jboss.jms.message.JBossTextMessage;
 import org.jboss.jms.server.MessagingTimeoutFactory;
@@ -577,6 +579,7 @@
 
       synchronized (lock)
       {
+         monitor.registerMessage(ref, null);
          messageRefs.addFirst(ref, ref.getMessage().getPriority());
       }
 
@@ -1064,7 +1067,13 @@
 
       public void afterRollback(boolean onePhase) throws Exception
       {
-      	//NOOP
+         //tx rolled back, we need to inform the monitor
+         for(Iterator i = refsToAdd.iterator(); i.hasNext(); )
+         {
+            log.error("* * * in roll back, telling the monitor about the change.");
+            MessageReference ref = (MessageReference)i.next();
+            monitor.unmarkSending(ref);
+         }
       }
 
       public String toString()

Modified: branches/Branch_JBMESSAGING_1416/src/main/org/jboss/messaging/core/impl/OrderingGroup.java
===================================================================
--- branches/Branch_JBMESSAGING_1416/src/main/org/jboss/messaging/core/impl/OrderingGroup.java	2008-10-23 22:05:06 UTC (rev 5177)
+++ branches/Branch_JBMESSAGING_1416/src/main/org/jboss/messaging/core/impl/OrderingGroup.java	2008-10-24 10:34:38 UTC (rev 5178)
@@ -160,20 +160,6 @@
    /**
     * @param ref
     */
-   public void releaseSend(MessageReference ref)
-   {
-      ReferenceHolder holder = sortedList.getFirst();
-      if (holder == null) return;
-      
-      if (holder.matchMessage(ref))
-      {
-         holder.releaseSend();
-      }
-   }
-
-   /**
-    * @param ref
-    */
    public void dropSend(MessageReference ref)
    {
       ReferenceHolder holder = sortedList.getFirst();
@@ -192,6 +178,19 @@
       }
    }
 
+   public void unmarkSending(MessageReference ref)
+   {
+      ReferenceHolder holder = sortedList.getFirst();
+      if (holder == null)
+      {
+         return;
+      }
+      if (holder.matchMessage(ref))
+      {
+         holder.unmarkSending();
+      }
+   }
+
    /**
     * @param ref
     */
@@ -323,8 +322,9 @@
    /**
     * @param ref2
     */
-   public long releaseSend()
+   public long unmarkSending()
    {
+System.err.println("* * * unmarking ref : " + ref);
       if (pendingSentCount > 0)
       {
          pendingSentCount--;

Modified: branches/Branch_JBMESSAGING_1416/src/main/org/jboss/messaging/core/impl/OrderingGroupMonitor.java
===================================================================
--- branches/Branch_JBMESSAGING_1416/src/main/org/jboss/messaging/core/impl/OrderingGroupMonitor.java	2008-10-23 22:05:06 UTC (rev 5177)
+++ branches/Branch_JBMESSAGING_1416/src/main/org/jboss/messaging/core/impl/OrderingGroupMonitor.java	2008-10-24 10:34:38 UTC (rev 5178)
@@ -182,6 +182,26 @@
    }
 
    /**
+    * reducing the refcount, if zero, remove it.
+    */
+   public void unmarkSending(MessageReference ref)
+   {
+      String grpName = extractGroupName(ref);
+      if (grpName == null)
+      {
+         return;
+      }
+      synchronized (orderingGroups)
+      {
+         OrderingGroup group = orderingGroups.get(grpName);
+         if (group != null)
+         {
+            group.unmarkSending(ref);
+         }
+      }
+   }
+
+   /**
     * @param ref
     */
    public void markSending(MessageReference ref)