Fix the concurrency issue for the endpoint operation invocation

git-svn-id: http://svn.us.apache.org/repos/asf/tuscany@1145738 13f79535-47bb-0310-9956-ffa450edef68

3 files changed, 37 insertions, 19 deletions

diff --git a/sca-java-2.x/trunk/modules/core/src/main/java/org/apache/tuscany/sca/core/invocation/impl/AsyncJDKInvocationHandler.java b/sca-java-2.x/trunk/modules/core/src/main/java/org/apache/tuscany/sca/core/invocation/impl/AsyncJDKInvocationHandler.java
index e46b2b7046..29d2da1487 100644
--- a/sca-java-2.x/trunk/modules/core/src/main/java/org/apache/tuscany/sca/core/invocation/impl/AsyncJDKInvocationHandler.java
+++ b/sca-java-2.x/trunk/modules/core/src/main/java/org/apache/tuscany/sca/core/invocation/impl/AsyncJDKInvocationHandler.java
@@ -29,7 +29,6 @@ import java.security.AccessController;
 import java.security.PrivilegedAction;
 import java.util.Arrays;
 import java.util.List;
-import java.util.Map;
 import java.util.concurrent.Callable;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ExecutorService;
@@ -66,7 +65,6 @@ import org.apache.tuscany.sca.core.UtilityExtensionPoint;
 import org.apache.tuscany.sca.core.assembly.RuntimeAssemblyFactory;
 import org.apache.tuscany.sca.core.invocation.AsyncFaultWrapper;
 import org.apache.tuscany.sca.core.invocation.AsyncResponseException;
-import org.apache.tuscany.sca.core.invocation.AsyncResponseHandler;
 import org.apache.tuscany.sca.core.invocation.AsyncResponseService;
 import org.apache.tuscany.sca.core.invocation.JDKAsyncResponseInvoker;
 import org.apache.tuscany.sca.interfacedef.InvalidInterfaceException;
@@ -373,12 +371,12 @@ public class AsyncJDKInvocationHandler extends JDKInvocationHandler {
                 		invokeAsync(chain, args, invocable, future.getUniqueID());
                 	} else {
                 		// Binding does not support native async invocations
-                		invoke(chain, args, invocable, future.getUniqueID());
+                		invoke(asyncMethod, chain, args, invocable, future.getUniqueID());
                 	} // end if
                     // The result is returned asynchronously via the future...
                 } else {
                     // ... the service is synchronous ...
-                    result = invoke(chain, args, invocable);
+                    result = invoke(asyncMethod, chain, args, invocable);
                     Type type = null;
                     if (asyncMethod.getReturnType() == Future.class) {
                         // For callback async method, where a Future is returned
diff --git a/sca-java-2.x/trunk/modules/core/src/main/java/org/apache/tuscany/sca/core/invocation/impl/JDKCallbackInvocationHandler.java b/sca-java-2.x/trunk/modules/core/src/main/java/org/apache/tuscany/sca/core/invocation/impl/JDKCallbackInvocationHandler.java
index 9b51aefe39..c35b70ad74 100644
--- a/sca-java-2.x/trunk/modules/core/src/main/java/org/apache/tuscany/sca/core/invocation/impl/JDKCallbackInvocationHandler.java
+++ b/sca-java-2.x/trunk/modules/core/src/main/java/org/apache/tuscany/sca/core/invocation/impl/JDKCallbackInvocationHandler.java
@@ -30,6 +30,7 @@ import org.apache.tuscany.sca.invocation.Invoker;
 import org.apache.tuscany.sca.invocation.Message;
 import org.apache.tuscany.sca.invocation.MessageFactory;
 import org.apache.tuscany.sca.runtime.Invocable;
+import org.apache.tuscany.sca.runtime.RuntimeEndpoint;
 import org.apache.tuscany.sca.runtime.RuntimeEndpointReference;
 import org.oasisopen.sca.ServiceReference;
 import org.oasisopen.sca.ServiceRuntimeException;
@@ -72,7 +73,7 @@ public class JDKCallbackInvocationHandler extends JDKInvocationHandler {
 
         try {
         	String msgID = ((CallbackServiceReferenceImpl)callableReference).getMsgID();
-            return invoke(chain, args, wire, msgID );
+            return invoke(method, chain, args, wire, msgID );
         } catch (InvocationTargetException e) {
             Throwable t = e.getCause();
             throw t;
@@ -91,7 +92,7 @@ public class JDKCallbackInvocationHandler extends JDKInvocationHandler {
      * @throws Throwable - if any exception occurs during the invocation
      */
     @Override
-    protected Object invoke(InvocationChain chain, Object[] args, Invocable source, String msgID)
+    protected Object invoke(Method method, InvocationChain chain, Object[] args, Invocable source, String msgID)
                          throws Throwable {
         Message msg = messageFactory.createMessage();
         if (source instanceof RuntimeEndpointReference) {
@@ -105,7 +106,19 @@ public class JDKCallbackInvocationHandler extends JDKInvocationHandler {
             }
         }
         Invoker headInvoker = chain.getHeadInvoker();
-        Operation operation = chain.getTargetOperation();
+        
+        Operation operation = null;
+        if(source instanceof RuntimeEndpoint) {
+            for (InvocationChain c : source.getInvocationChains()) {
+                Operation op = c.getTargetOperation();
+                if (method.getName().equals(op.getName())) {
+                    operation = op;
+                    break;
+                }
+            }
+        } else {
+            operation = chain.getTargetOperation();
+        }
         msg.setOperation(operation);
         msg.setBody(args);
 
diff --git a/sca-java-2.x/trunk/modules/core/src/main/java/org/apache/tuscany/sca/core/invocation/impl/JDKInvocationHandler.java b/sca-java-2.x/trunk/modules/core/src/main/java/org/apache/tuscany/sca/core/invocation/impl/JDKInvocationHandler.java
index 5bb2354520..41f469cc1b 100644
--- a/sca-java-2.x/trunk/modules/core/src/main/java/org/apache/tuscany/sca/core/invocation/impl/JDKInvocationHandler.java
+++ b/sca-java-2.x/trunk/modules/core/src/main/java/org/apache/tuscany/sca/core/invocation/impl/JDKInvocationHandler.java
@@ -122,7 +122,7 @@ public class JDKInvocationHandler implements InvocationHandler, Serializable {
             promotedArgs = removeOutOnlyArgs(sourceOp, promotedArgs );
         } 
         
-        Object result = invoke(chain, promotedArgs, source);
+        Object result = invoke(method, chain, promotedArgs, source);
         
         // TODO - Based on the code in JavaInterfaceIntrospectorImpl, it seems there are
         // some cases involving generics that we're not taking into account.
@@ -239,13 +239,7 @@ public class JDKInvocationHandler implements InvocationHandler, Serializable {
 
     protected synchronized InvocationChain getInvocationChain(Method method, Invocable source) {
         if (source instanceof RuntimeEndpoint) {
-            InvocationChain invocationChain = source.getBindingInvocationChain();
-            for (InvocationChain chain : source.getInvocationChains()) {
-                Operation operation = chain.getTargetOperation();
-                if (method.getName().equals(operation.getName())) {
-                    invocationChain.setTargetOperation(operation);
-                }
-            }
+            // [rfeng] Start with the binding invocation chain
             return source.getBindingInvocationChain();
         }
         if (fixedWire && chains.containsKey(method)) {
@@ -273,9 +267,9 @@ public class JDKInvocationHandler implements InvocationHandler, Serializable {
         this.target = endpoint;
     }
     
-    protected Object invoke(InvocationChain chain, Object[] args, Invocable source)
+    protected Object invoke(Method method, InvocationChain chain, Object[] args, Invocable source)
                             throws Throwable {
-    	return invoke( chain, args, source, null );
+    	return invoke( method, chain, args, source, null );
     }
 
     /**
@@ -287,7 +281,7 @@ public class JDKInvocationHandler implements InvocationHandler, Serializable {
      * @return - the Response message from the invocation
      * @throws Throwable - if any exception occurs during the invocation
      */
-    protected Object invoke(InvocationChain chain, Object[] args, Invocable source, String msgID)
+    protected Object invoke(Method method, InvocationChain chain, Object[] args, Invocable source, String msgID)
                          throws Throwable {
         Message msg = messageFactory.createMessage();
         if (source instanceof RuntimeEndpointReference) {
@@ -301,7 +295,20 @@ public class JDKInvocationHandler implements InvocationHandler, Serializable {
             }
         }
         Invoker headInvoker = chain.getHeadInvoker();
-        Operation operation = chain.getTargetOperation();
+        Operation operation = null;
+        if(source instanceof RuntimeEndpoint) {
+            // [rfeng] We cannot use the targetOperation from the binding invocation chain.
+            // For each method, we need to find the matching operation so that we can set the operation on to the message
+            for (InvocationChain c : source.getInvocationChains()) {
+                Operation op = c.getTargetOperation();
+                if (method.getName().equals(op.getName())) {
+                    operation = op;
+                    break;
+                }
+            }
+        } else {
+            operation = chain.getTargetOperation();
+        }
         msg.setOperation(operation);
         msg.setBody(args);