[jboss-svn-commits] JBL Code SVN: r25241 - in labs/jbossesb/trunk/product/rosetta/src/org/jboss/internal/soa/esb: couriers and 1 other directory.

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

Author: tfennelly
Date: 2009-02-12 09:05:17 -0500 (Thu, 12 Feb 2009)
New Revision: 25241

Modified:
   labs/jbossesb/trunk/product/rosetta/src/org/jboss/internal/soa/esb/addressing/eprs/DefaultInVMReplyToEpr.java
   labs/jbossesb/trunk/product/rosetta/src/org/jboss/internal/soa/esb/couriers/InVMCourier.java
Log:
https://jira.jboss.org/jira/browse/JBESB-2398

Modified: labs/jbossesb/trunk/product/rosetta/src/org/jboss/internal/soa/esb/addressing/eprs/DefaultInVMReplyToEpr.java
===================================================================
--- labs/jbossesb/trunk/product/rosetta/src/org/jboss/internal/soa/esb/addressing/eprs/DefaultInVMReplyToEpr.java	2009-02-12 09:59:56 UTC (rev 25240)
+++ labs/jbossesb/trunk/product/rosetta/src/org/jboss/internal/soa/esb/addressing/eprs/DefaultInVMReplyToEpr.java	2009-02-12 14:05:17 UTC (rev 25241)
@@ -19,7 +19,6 @@
 
 package org.jboss.internal.soa.esb.addressing.eprs;
 
-import org.jboss.internal.soa.esb.couriers.InVMCourier;
 import org.jboss.soa.esb.addressing.EPR;
 import org.jboss.soa.esb.addressing.eprs.InVMEpr;
 import org.jboss.soa.esb.couriers.CourierFactory;
@@ -29,7 +28,7 @@
 
 public class DefaultInVMReplyToEpr extends InVMEpr {
 
-    private static ThreadLocal<InVMEpr> invmReplyToEPRTL = new ThreadLocal<InVMEpr>();
+    private static ThreadLocal<InVMEPRThreadTracker> invmThreadReplyTracker = new ThreadLocal<InVMEPRThreadTracker>();
 
     public DefaultInVMReplyToEpr(InVMEpr epr) {
         super(getThreadEPR());
@@ -38,23 +37,32 @@
     private static EPR getThreadEPR() {
         // We assign an InVMEpr/Courier to each thread
 
-        InVMEpr threadEPR = invmReplyToEPRTL.get();
+        InVMEPRThreadTracker tracker = invmThreadReplyTracker.get();
 
-        if(threadEPR == null) {
-            try {
-                threadEPR = new InVMEpr(URI.create("invm://thread-" + Thread.currentThread().getId()));
-                invmReplyToEPRTL.set(threadEPR);
-                CourierFactory.getInstance().addInVMCourier(threadEPR);
-            } catch (URISyntaxException e) {
-                throw new RuntimeException("Unexpected Exception creating replyTo InVMEpr.", e);
-            }
+        if(tracker == null) {
+            tracker = new InVMEPRThreadTracker();
+            invmThreadReplyTracker.set(tracker);
+        } else if(tracker.lastEPR != null) {
+            // Deactivate the last InVMCourier used on this thread...
+            CourierFactory.getInstance().removeInVMCourier(tracker.lastEPR);
+            tracker.lastEPR = null;
         }
 
-        // Rest the courier in case there are stale messages there
-        // waiting to be picked up...
-        InVMCourier threadCourier = CourierFactory.getInstance().getInVMCourier(threadEPR);
-        threadCourier.reset();
+        try {
+            // Create a brand new EPR for this invocation and register a
+            // new InVMCourier for it...
+            tracker.counter++;
+            tracker.lastEPR = new InVMEpr(URI.create("invm://thread-" + Thread.currentThread().getId() + "-" + tracker.counter));
+            CourierFactory.getInstance().addInVMCourier(tracker.lastEPR);
+        } catch (URISyntaxException e) {
+            throw new RuntimeException("Unexpected Exception creating replyTo InVMEpr.", e);
+        }
 
-        return threadEPR;
+        return tracker.lastEPR;
     }
+
+    private static class InVMEPRThreadTracker {
+        private InVMEpr lastEPR;
+        private long counter = 0;
+    }
 }
\ No newline at end of file

Modified: labs/jbossesb/trunk/product/rosetta/src/org/jboss/internal/soa/esb/couriers/InVMCourier.java
===================================================================
--- labs/jbossesb/trunk/product/rosetta/src/org/jboss/internal/soa/esb/couriers/InVMCourier.java	2009-02-12 09:59:56 UTC (rev 25240)
+++ labs/jbossesb/trunk/product/rosetta/src/org/jboss/internal/soa/esb/couriers/InVMCourier.java	2009-02-12 14:05:17 UTC (rev 25241)
@@ -336,6 +336,11 @@
 
     public void setActive(boolean active) {
         isActive = active;
+
+        if(!active && !messageQueue.isEmpty()) {
+            logger.error("Deactivated non-empty InVM Queue.  Undelivered messages will be lost.");
+            messageQueue.clear();
+        }
     }
 
     public void clean() {