[jboss-svn-commits] JBL Code SVN: r26615 - in labs/jbossesb/branches/JBESB_4_4_GA_CP/product/rosetta/src/org/jboss: soa/esb/client and 1 other directory.

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

Author: kevin.conner at jboss.com
Date: 2009-05-19 07:32:27 -0400 (Tue, 19 May 2009)
New Revision: 26615

Modified:
   labs/jbossesb/branches/JBESB_4_4_GA_CP/product/rosetta/src/org/jboss/internal/soa/esb/addressing/eprs/DefaultInVMReplyToEpr.java
   labs/jbossesb/branches/JBESB_4_4_GA_CP/product/rosetta/src/org/jboss/soa/esb/client/ServiceInvoker.java
Log:
Pulled across 'special' cleanup of InVM couriers: JBESB-2523

Modified: labs/jbossesb/branches/JBESB_4_4_GA_CP/product/rosetta/src/org/jboss/internal/soa/esb/addressing/eprs/DefaultInVMReplyToEpr.java
===================================================================
--- labs/jbossesb/branches/JBESB_4_4_GA_CP/product/rosetta/src/org/jboss/internal/soa/esb/addressing/eprs/DefaultInVMReplyToEpr.java	2009-05-19 11:21:37 UTC (rev 26614)
+++ labs/jbossesb/branches/JBESB_4_4_GA_CP/product/rosetta/src/org/jboss/internal/soa/esb/addressing/eprs/DefaultInVMReplyToEpr.java	2009-05-19 11:32:27 UTC (rev 26615)
@@ -28,41 +28,25 @@
 
 public class DefaultInVMReplyToEpr extends InVMEpr {
 
-    private static ThreadLocal<InVMEPRThreadTracker> invmThreadReplyTracker = new ThreadLocal<InVMEPRThreadTracker>();
+    private static volatile long counter = 0;
 
     public DefaultInVMReplyToEpr(InVMEpr epr) {
-        super(getThreadEPR());
+        super(createEprAndCourier());
     }
 
-    private static EPR getThreadEPR() {
-        // We assign an InVMEpr/Courier to each thread
+    private static EPR createEprAndCourier() {
+        try {
+            InVMEpr epr;
 
-        InVMEPRThreadTracker tracker = invmThreadReplyTracker.get();
+            synchronized (DefaultInVMReplyToEpr.class) {
+                counter++;
+                epr = new InVMEpr(URI.create("invm://thread-" + Thread.currentThread().getId() + "-" + counter));
+            }
+            CourierFactory.getInstance().addInVMCourier(epr);
 
-        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;
-        }
-
-        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);
+            return epr;
         } catch (URISyntaxException e) {
             throw new RuntimeException("Unexpected Exception creating replyTo InVMEpr.", e);
         }
-
-        return tracker.lastEPR;
     }
-
-    private static class InVMEPRThreadTracker {
-        private InVMEpr lastEPR;
-        private long counter = 0;
-    }
 }
\ No newline at end of file

Modified: labs/jbossesb/branches/JBESB_4_4_GA_CP/product/rosetta/src/org/jboss/soa/esb/client/ServiceInvoker.java
===================================================================
--- labs/jbossesb/branches/JBESB_4_4_GA_CP/product/rosetta/src/org/jboss/soa/esb/client/ServiceInvoker.java	2009-05-19 11:21:37 UTC (rev 26614)
+++ labs/jbossesb/branches/JBESB_4_4_GA_CP/product/rosetta/src/org/jboss/soa/esb/client/ServiceInvoker.java	2009-05-19 11:32:27 UTC (rev 26615)
@@ -34,6 +34,7 @@
 import org.jboss.soa.esb.addressing.EPR;
 import org.jboss.soa.esb.addressing.MalformedEPRException;
 import org.jboss.soa.esb.addressing.PortReference;
+import org.jboss.soa.esb.addressing.eprs.InVMEpr;
 import org.jboss.soa.esb.addressing.util.DefaultReplyTo;
 import org.jboss.soa.esb.common.Configuration;
 import org.jboss.soa.esb.common.Environment;
@@ -622,12 +623,24 @@
 
                     throw new MessageDeliverException("Caught unexpected throwable during send. Bailing-out!", t);
                 } finally {
-                    CourierUtil.cleanCourier(courier);
+                    EPR replyToEpr = null;
+                    
+                    try {
+                        CourierUtil.cleanCourier(courier);
+                        // put back the old To since we will have changed it.
 
-                    // put back the old To since we will have changed it.
-
-                    call.setTo(currentToEpr) ;
-                    call.setReplyTo(currentReplyToEpr) ;
+                        replyToEpr = call.getReplyTo();
+                        call.setTo(currentToEpr);
+                        call.setReplyTo(currentReplyToEpr);
+                    } finally {
+                        // Need to do "special" cleanup for InVM replyTo couriers.  Of course, this may
+                        // not be most "pure" to the courier architecture, but we can't use Courier.cleanup because it
+                        // is getting called multiple times between async delivery and pickup, which totally breaks
+                        // InVM delivery because it zaps the in-mem queue.
+                        if (replyToEpr instanceof InVMEpr) {
+                            CourierFactory.getInstance().removeInVMCourier((InVMEpr) replyToEpr);
+                        }
+                    }
                 }
             }