Enhance the JAX-WS async client support and fix couple of issues

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

5 files changed, 392 insertions, 285 deletions

diff --git a/sca-java-2.x/trunk/modules/core/src/main/java/org/apache/tuscany/sca/core/invocation/DefaultProxyFactoryExtensionPoint.java b/sca-java-2.x/trunk/modules/core/src/main/java/org/apache/tuscany/sca/core/invocation/DefaultProxyFactoryExtensionPoint.java
index af6f039775..03505da302 100644
--- a/sca-java-2.x/trunk/modules/core/src/main/java/org/apache/tuscany/sca/core/invocation/DefaultProxyFactoryExtensionPoint.java
+++ b/sca-java-2.x/trunk/modules/core/src/main/java/org/apache/tuscany/sca/core/invocation/DefaultProxyFactoryExtensionPoint.java
@@ -39,21 +39,21 @@ public class DefaultProxyFactoryExtensionPoint implements ProxyFactoryExtensionP
     private ProxyFactory interfaceFactory;
     private ProxyFactory classFactory;
 
-    public DefaultProxyFactoryExtensionPoint(ExtensionPointRegistry extensionPoints) {
-        UtilityExtensionPoint utilities = extensionPoints.getExtensionPoint(UtilityExtensionPoint.class);
+    public DefaultProxyFactoryExtensionPoint(ExtensionPointRegistry registry) {
+        UtilityExtensionPoint utilities = registry.getExtensionPoint(UtilityExtensionPoint.class);
         this.interfaceContractMapper = utilities.getUtility(InterfaceContractMapper.class);
 
-        FactoryExtensionPoint modelFactories = extensionPoints.getExtensionPoint(FactoryExtensionPoint.class);
+        FactoryExtensionPoint modelFactories = registry.getExtensionPoint(FactoryExtensionPoint.class);
         this.messageFactory = modelFactories.getFactory(MessageFactory.class);
 
-        interfaceFactory = new JDKProxyFactory(messageFactory, interfaceContractMapper);
+        interfaceFactory = new JDKProxyFactory(registry, messageFactory, interfaceContractMapper);
     }
 
-    public DefaultProxyFactoryExtensionPoint(MessageFactory messageFactory, InterfaceContractMapper mapper) {
-        this.interfaceContractMapper = mapper;
-        this.messageFactory = messageFactory;
-        interfaceFactory = new JDKProxyFactory(messageFactory, mapper);
-    }
+    //    public DefaultProxyFactoryExtensionPoint(MessageFactory messageFactory, InterfaceContractMapper mapper) {
+    //        this.interfaceContractMapper = mapper;
+    //        this.messageFactory = messageFactory;
+    //        interfaceFactory = new JDKProxyFactory(null, messageFactory, mapper);
+    //    }
 
     public ProxyFactory getClassProxyFactory() {
         return classFactory;
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 4371399f58..4af9ce585b 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
@@ -19,18 +19,21 @@
 
 package org.apache.tuscany.sca.core.invocation.impl;
 
+import java.beans.Introspector;
+import java.beans.PropertyDescriptor;
 import java.io.StringReader;
 import java.lang.reflect.Method;
+import java.lang.reflect.ParameterizedType;
+import java.lang.reflect.Type;
 import java.security.AccessController;
 import java.security.PrivilegedAction;
 import java.util.Arrays;
 import java.util.List;
-import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.Callable;
 import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Future;
-import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
 
 import javax.xml.namespace.QName;
 import javax.xml.stream.XMLStreamException;
@@ -56,6 +59,7 @@ import org.apache.tuscany.sca.contribution.processor.StAXArtifactProcessorExtens
 import org.apache.tuscany.sca.contribution.processor.ValidatingXMLInputFactory;
 import org.apache.tuscany.sca.core.ExtensionPointRegistry;
 import org.apache.tuscany.sca.core.FactoryExtensionPoint;
+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;
@@ -64,6 +68,7 @@ import org.apache.tuscany.sca.interfacedef.InvalidInterfaceException;
 import org.apache.tuscany.sca.interfacedef.java.JavaInterfaceContract;
 import org.apache.tuscany.sca.interfacedef.java.JavaInterfaceFactory;
 import org.apache.tuscany.sca.interfacedef.util.FaultException;
+import org.apache.tuscany.sca.interfacedef.util.WrapperInfo;
 import org.apache.tuscany.sca.invocation.InvocationChain;
 import org.apache.tuscany.sca.invocation.MessageFactory;
 import org.apache.tuscany.sca.policy.Intent;
@@ -73,8 +78,9 @@ import org.apache.tuscany.sca.runtime.Invocable;
 import org.apache.tuscany.sca.runtime.RuntimeComponent;
 import org.apache.tuscany.sca.runtime.RuntimeEndpoint;
 import org.apache.tuscany.sca.runtime.RuntimeEndpointReference;
+import org.apache.tuscany.sca.work.WorkScheduler;
 import org.oasisopen.sca.ServiceReference;
-import org.oasisopen.sca.ServiceRuntimeException;  
+import org.oasisopen.sca.ServiceRuntimeException;
 
 /**
  * An InvocationHandler which deals with JAXWS-defined asynchronous client Java API method calls
@@ -92,26 +98,46 @@ import org.oasisopen.sca.ServiceRuntimeException;
  *   a callback operation that is automatically called when the async call completes
  */
 public class AsyncJDKInvocationHandler extends JDKInvocationHandler {
-    
+
     private static final long serialVersionUID = 1L;
-    
-	private static int invocationCount = 10;	// # of threads to use
-	private static long maxWaitTime = 30;	    // Max wait time for completion = 30sec
-	
-	// Run the async service invocations using a ThreadPoolExecutor
-	private static ThreadPoolExecutor theExecutor = new ThreadPoolExecutor( invocationCount, invocationCount,
-												                maxWaitTime, TimeUnit.SECONDS,
-												                new ArrayBlockingQueue<Runnable>( invocationCount ) );
-	
-
-    public AsyncJDKInvocationHandler(MessageFactory messageFactory, ServiceReference<?> callableReference) {
+
+    private static int invocationCount = 10; // # of threads to use
+    private static long maxWaitTime = 30; // Max wait time for completion = 30sec
+
+    // Run the async service invocations using a ThreadPoolExecutor
+    private ExecutorService theExecutor;
+
+    public AsyncJDKInvocationHandler(ExtensionPointRegistry registry,
+                                     MessageFactory messageFactory,
+                                     ServiceReference<?> callableReference) {
         super(messageFactory, callableReference);
+        initExecutorService(registry);
     }
 
-    public AsyncJDKInvocationHandler(MessageFactory messageFactory,
+    public AsyncJDKInvocationHandler(ExtensionPointRegistry registry,
+                                     MessageFactory messageFactory,
                                      Class<?> businessInterface,
                                      Invocable source) {
         super(messageFactory, businessInterface, source);
+        initExecutorService(registry);
+    }
+
+    private final void initExecutorService(ExtensionPointRegistry registry) {
+        UtilityExtensionPoint utilities = registry.getExtensionPoint(UtilityExtensionPoint.class);
+        WorkScheduler scheduler = utilities.getUtility(WorkScheduler.class);
+        theExecutor = scheduler.getExecutorService();
+
+        /*
+        synchronized (AsyncJDKInvocationHandler.class) {
+            theExecutor = utilities.getUtility(ExecutorService.class);
+            if (theExecutor == null) {
+                theExecutor =
+                    new ThreadPoolExecutor(invocationCount, invocationCount, maxWaitTime, TimeUnit.SECONDS,
+                                           new ArrayBlockingQueue<Runnable>(invocationCount));
+                utilities.addUtility(ExecutorService.class, theExecutor);
+            }
+        }
+        */
     }
 
     /**
@@ -120,17 +146,17 @@ public class AsyncJDKInvocationHandler extends JDKInvocationHandler {
      */
     @Override
     public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
-        
+
         // force the bind of the reference so that we can look at the 
         // target contract to see if it's asynchronous 
         source.getInvocationChains();
-        
+
         if (isAsyncCallback(method)) {
-            return doInvokeAsyncCallback(proxy, method, args);            
+            return doInvokeAsyncCallback(proxy, method, args);
         } else if (isAsyncPoll(method)) {
-            return doInvokeAsyncPoll(proxy, method, args);            
+            return doInvokeAsyncPoll(proxy, method, args);
         } else {
-        	// Regular synchronous method call
+            // Regular synchronous method call
             return doInvokeSync(proxy, method, args);
         }
     }
@@ -141,9 +167,9 @@ public class AsyncJDKInvocationHandler extends JDKInvocationHandler {
      * @return - true if the method has the form of an async callback
      */
     protected boolean isAsyncCallback(Method method) {
-        if (method.getName().endsWith("Async") && (method.getReturnType().isAssignableFrom(Future.class))) {
+        if (method.getName().endsWith("Async") && (method.getReturnType() == Future.class)) {
             if (method.getParameterTypes().length > 0) {
-                return method.getParameterTypes()[method.getParameterTypes().length-1].isAssignableFrom(AsyncHandler.class);
+                return method.getParameterTypes()[method.getParameterTypes().length - 1] == AsyncHandler.class;
             }
         }
         return false;
@@ -155,7 +181,7 @@ public class AsyncJDKInvocationHandler extends JDKInvocationHandler {
      * @return - true if the method has the form of an async polling method
      */
     protected boolean isAsyncPoll(Method method) {
-        return method.getName().endsWith("Async") && (method.getReturnType().isAssignableFrom(Response.class));
+        return method.getName().endsWith("Async") && (method.getReturnType() == Response.class);
     }
 
     /**
@@ -167,46 +193,50 @@ public class AsyncJDKInvocationHandler extends JDKInvocationHandler {
      *           type of the response
      */
     @SuppressWarnings("unchecked")
-	protected Response doInvokeAsyncPoll(Object proxy, Method asyncMethod, Object[] args) {
-        Class<?> returnType = getNonAsyncMethod(asyncMethod).getReturnType();
+    protected Response doInvokeAsyncPoll(Object proxy, Method asyncMethod, Object[] args) {
+        Method method = getNonAsyncMethod(asyncMethod);
+        Class<?> returnType = method.getReturnType();
         // Allocate the Future<?> / Response<?> object - note: Response<?> is a subclass of Future<?>
-        AsyncInvocationFutureImpl future = AsyncInvocationFutureImpl.newInstance( returnType, getInterfaceClassloader() );
+        AsyncInvocationFutureImpl future = AsyncInvocationFutureImpl.newInstance(returnType, getInterfaceClassloader());
         try {
-            invokeAsync(proxy, getNonAsyncMethod(asyncMethod), args, future);
+            invokeAsync(proxy, method, args, future, asyncMethod);
         } catch (Exception e) {
-            future.setFault( new AsyncFaultWrapper(e) );
-        } catch (Throwable t ) {
-        	Exception e = new ServiceRuntimeException("Received Throwable: " + t.getClass().getName() + 
-        			                                  " when invoking: " + asyncMethod.getName(), t);
-        	future.setFault( new AsyncFaultWrapper(e) );
+            future.setFault(new AsyncFaultWrapper(e));
+        } catch (Throwable t) {
+            Exception e =
+                new ServiceRuntimeException("Received Throwable: " + t.getClass().getName()
+                    + " when invoking: "
+                    + asyncMethod.getName(), t);
+            future.setFault(new AsyncFaultWrapper(e));
         } // end try 
         return future;
     } // end method doInvokeAsyncPoll
-    
+
     /**
      * Provide a synchronous invocation of a service operation that is either synchronous or asynchronous
      * @return
      */
     protected Object doInvokeSync(Object proxy, Method method, Object[] args) throws Throwable {
-    	if ( isAsyncInvocation( source ) ) {
-    		// Target service is asynchronous
-    		Class<?> returnType = method.getReturnType();
-            AsyncInvocationFutureImpl future = AsyncInvocationFutureImpl.newInstance( returnType, getInterfaceClassloader() );
-            invokeAsync(proxy, method, args, future);
+        if (isAsyncInvocation(source)) {
+            // Target service is asynchronous
+            Class<?> returnType = method.getReturnType();
+            AsyncInvocationFutureImpl future =
+                AsyncInvocationFutureImpl.newInstance(returnType, getInterfaceClassloader());
+            invokeAsync(proxy, method, args, future, method);
             // Wait for some maximum time for the result - 1000 seconds here
             // Really, if the service is async, the client should use async client methods to invoke the service
             // - and be prepared to wait a *really* long time
             Object response = null;
             try {
                 response = future.get(1000, TimeUnit.SECONDS);
-            } catch(ExecutionException ex) {
+            } catch (ExecutionException ex) {
                 throw ex.getCause();
             }
             return response;
-    	} else {
-    		// Target service is not asynchronous, so perform sync invocation
-    		return super.invoke(proxy, method, args);
-    	} // end if
+        } else {
+            // Target service is not asynchronous, so perform sync invocation
+            return super.invoke(proxy, method, args);
+        } // end if
     } // end method doInvokeSync
 
     /**
@@ -219,15 +249,23 @@ public class AsyncJDKInvocationHandler extends JDKInvocationHandler {
      *           the response
      */
     @SuppressWarnings("unchecked")
-	private Object doInvokeAsyncCallback(Object proxy, Method asyncMethod, Object[] args) {
-        AsyncHandler handler = (AsyncHandler)args[args.length-1];
-        Response response = doInvokeAsyncPoll(proxy,asyncMethod,Arrays.copyOf(args, args.length-1));
-        // Invoke the callback handler, if present
-        if( handler != null ) {
-        	handler.handleResponse(response);
-        } // end if
-        
-        return response;
+    private Object doInvokeAsyncCallback(final Object proxy, final Method asyncMethod, final Object[] args)
+        throws Exception {
+        Future<Response> future = theExecutor.submit(new Callable<Response>() {
+
+            @Override
+            public Response call() {
+                AsyncHandler handler = (AsyncHandler)args[args.length - 1];
+                Response response = doInvokeAsyncPoll(proxy, asyncMethod, Arrays.copyOf(args, args.length - 1));
+                // Invoke the callback handler, if present
+                if (handler != null) {
+                    handler.handleResponse(response);
+                } // end if
+                return response;
+            }
+        });
+        return future.get();
+
     } // end method doInvokeAsyncCallback
 
     /**
@@ -240,11 +278,15 @@ public class AsyncJDKInvocationHandler extends JDKInvocationHandler {
      * @throws Throwable - if an exception is thrown during the invocation
      */
     @SuppressWarnings("unchecked")
-	private void invokeAsync(Object proxy, Method method, Object[] args, AsyncInvocationFutureImpl future) throws Throwable {
+    private void invokeAsync(Object proxy,
+                             Method method,
+                             Object[] args,
+                             AsyncInvocationFutureImpl future,
+                             Method asyncMethod) throws Throwable {
         if (source == null) {
             throw new ServiceRuntimeException("No runtime source is available");
         }
-        
+
         if (source instanceof RuntimeEndpointReference) {
             RuntimeEndpointReference epr = (RuntimeEndpointReference)source;
             if (epr.isOutOfDate()) {
@@ -252,131 +294,173 @@ public class AsyncJDKInvocationHandler extends JDKInvocationHandler {
                 chains.clear();
             }
         } // end if
-        
+
         InvocationChain chain = getInvocationChain(method, source);
-        
+
         if (chain == null) {
             throw new IllegalArgumentException("No matching operation is found: " + method);
         }
-        
+
         // Organize for an async service
-        RuntimeEndpoint theEndpoint = getAsyncCallback( source );
+        RuntimeEndpoint theEndpoint = getAsyncCallback(source);
         boolean isAsyncService = false;
-        if( theEndpoint != null ) {
-        	// ... the service is asynchronous ...
-        	attachFuture( theEndpoint, future );
-        	isAsyncService = true;
+        if (theEndpoint != null) {
+            // ... the service is asynchronous ...
+            attachFuture(theEndpoint, future);
+            isAsyncService = true;
         } else {
-        	// ... the service is synchronous ...
+            // ... the service is synchronous ...
         } // end if
-        
-		// Perform the invocations on separate thread...
-		theExecutor.execute( new separateThreadInvoker( chain, args, source, future, isAsyncService ) );
+
+        // Perform the invocations on separate thread...
+        theExecutor.submit(new separateThreadInvoker(chain, args, source, future, asyncMethod, isAsyncService));
 
         return;
     } // end method invokeAsync
-    
-	/**
-	 * An inner class which acts as a runnable task for invoking services asynchronously on threads that are separate from
-	 * those used to execute operations of components
-	 * 
-	 * This supports both synchronous services and asynchronous services
-	 */
-	private class separateThreadInvoker implements Runnable {
-		
-		private AsyncInvocationFutureImpl future;
-		private InvocationChain chain;
-		private Object[] args;
-		private Invocable invocable;
-		private boolean isAsyncService;
-				
-		public separateThreadInvoker( InvocationChain chain, Object[] args, Invocable invocable,
-				                      AsyncInvocationFutureImpl future, boolean isAsyncService ) {
-			super();
-			this.chain = chain;
-			this.args = args;
-			this.invocable = invocable;
-			this.future = future;
-			this.isAsyncService = isAsyncService;
-		} // end constructor
-
-		public void run() {
-			Object result;
-			
-			try {
-				if( isAsyncService ) {
-		        	invoke(chain, args, invocable, future.getUniqueID());
-		        	// The result is returned asynchronously via the future...
-		        } else {
-		        	// ... the service is synchronous ...
-		        	result = invoke(chain, args, invocable);
-		        	future.setResponse(result);
-				} // end if
-			} catch ( ServiceRuntimeException s ) {
-				Throwable e = s.getCause();
-				if( e != null && e instanceof FaultException ) {
-					if( "AsyncResponse".equals(e.getMessage()) ) {
-						// Do nothing...
-					} else { 
-						future.setFault( new AsyncFaultWrapper( s ) );
-					} // end if 
-				} // end if
-            } catch ( AsyncResponseException ar ) {
+
+    /**
+     * An inner class which acts as a runnable task for invoking services asynchronously on threads that are separate from
+     * those used to execute operations of components
+     * 
+     * This supports both synchronous services and asynchronous services
+     */
+    private class separateThreadInvoker implements Runnable {
+
+        private AsyncInvocationFutureImpl future;
+        private Method asyncMethod;
+        private InvocationChain chain;
+        private Object[] args;
+        private Invocable invocable;
+        private boolean isAsyncService;
+
+        public separateThreadInvoker(InvocationChain chain,
+                                     Object[] args,
+                                     Invocable invocable,
+                                     AsyncInvocationFutureImpl future,
+                                     Method asyncMethod,
+                                     boolean isAsyncService) {
+            super();
+            this.chain = chain;
+            this.asyncMethod = asyncMethod;
+            this.args = args;
+            this.invocable = invocable;
+            this.future = future;
+            this.isAsyncService = isAsyncService;
+        } // end constructor
+
+        public void run() {
+            Object result;
+
+            try {
+                if (isAsyncService) {
+                    invoke(chain, args, invocable, future.getUniqueID());
+                    // The result is returned asynchronously via the future...
+                } else {
+                    // ... the service is synchronous ...
+                    result = invoke(chain, args, invocable);
+                    Type type = null;
+                    if (asyncMethod.getReturnType() == Future.class) {
+                        // For callback async menthod
+                        Type[] types = asyncMethod.getGenericParameterTypes();
+                        if (types.length > 0 && asyncMethod.getParameterTypes()[types.length - 1] == AsyncHandler.class) {
+                            // Last paremeter, AsyncHandler<T>
+                            type = types[types.length - 1];
+                        }
+                    } else if (asyncMethod.getReturnType() == Response.class) {
+                        // For the polling method, Response<T>
+                        type = asyncMethod.getGenericReturnType();
+                    }
+                    if (type instanceof ParameterizedType) {
+                        // Check if the parameterized type of Response<T> is a doc-lit-wrapper class
+                        Class<?> wrapperClass = (Class<?>)((ParameterizedType)type).getActualTypeArguments()[0];
+                        WrapperInfo wrapperInfo = chain.getSourceOperation().getWrapper();
+                        if (wrapperInfo != null && wrapperInfo.getOutputWrapperClass() == wrapperClass) {
+                            Object wrapper = wrapperClass.newInstance();
+                            // Find the 1st matching property
+                            for (PropertyDescriptor p : Introspector.getBeanInfo(wrapperClass).getPropertyDescriptors()) {
+                                if (p.getWriteMethod() == null) {
+                                    // There is a "class" property ...
+                                    continue;
+                                }
+                                if (p.getWriteMethod().getParameterTypes()[0].isInstance(result)) {
+                                    p.getWriteMethod().invoke(wrapper, result);
+                                    result = wrapper;
+                                    break;
+                                }
+                            }
+
+                        }
+                    }
+                    future.setResponse(result);
+                } // end if
+            } catch (ServiceRuntimeException s) {
+                Throwable e = s.getCause();
+                if (e != null && e instanceof FaultException) {
+                    if ("AsyncResponse".equals(e.getMessage())) {
+                        // Do nothing...
+                    } else {
+                        future.setFault(new AsyncFaultWrapper(s));
+                    } // end if 
+                } // end if
+            } catch (AsyncResponseException ar) {
                 // do nothing			
-			} catch ( Throwable t ) {
-				System.out.println("Async invoke got exception: " + t.toString());
-				future.setFault( new AsyncFaultWrapper( t ) );
-			} // end try
-			
-		} // end method run
-		
-	} // end class separateThreadInvoker
-    
+            } catch (Throwable t) {
+                System.out.println("Async invoke got exception: " + t.toString());
+                future.setFault(new AsyncFaultWrapper(t));
+            } // end try
+
+        } // end method run
+
+    } // end class separateThreadInvoker
+
     /**
      * Attaches a future to the callback endpoint - so that the Future is triggered when a response is
      * received from the asynchronous service invocation associated with the Future
      * @param endpoint - the async callback endpoint
      * @param future - the async invocation future to attach
      */
-    private void attachFuture( RuntimeEndpoint endpoint, AsyncInvocationFutureImpl<?> future ) {
-    	Implementation impl = endpoint.getComponent().getImplementation();
-    	AsyncResponseHandlerImpl<?> asyncHandler = (AsyncResponseHandlerImpl<?>) impl;
-    	asyncHandler.addFuture(future);
+    private void attachFuture(RuntimeEndpoint endpoint, AsyncInvocationFutureImpl<?> future) {
+        Implementation impl = endpoint.getComponent().getImplementation();
+        AsyncResponseHandlerImpl<?> asyncHandler = (AsyncResponseHandlerImpl<?>)impl;
+        asyncHandler.addFuture(future);
     } // end method attachFuture
-    
+
     /**
      * Get the async callback endpoint - if not already created, create and start it
      * @param source - the RuntimeEndpointReference which needs an async callback endpoint
      * @param future 
      * @return - the RuntimeEndpoint of the async callback
      */
-    private RuntimeEndpoint getAsyncCallback( Invocable source ) {
-    	if( !(source instanceof RuntimeEndpointReference) ) return null;
-		RuntimeEndpointReference epr = (RuntimeEndpointReference) source;
-    	if( !isAsyncInvocation( epr ) ) return null;
-    	RuntimeEndpoint endpoint;
-    	synchronized( epr ) {
-    		endpoint = (RuntimeEndpoint)epr.getCallbackEndpoint();
-    		// If the async callback endpoint is already created, return it...
-    		if( endpoint != null ) return endpoint;
-	    	// Create the endpoint for the async callback
-	    	endpoint = createAsyncCallbackEndpoint( epr );
-	    	epr.setCallbackEndpoint(endpoint);
-    	} // end synchronized
-    	
-    	// Activate the new callback endpoint
-    	startEndpoint( epr.getCompositeContext(), endpoint );
-    	endpoint.getInvocationChains();
-    	
-    	return endpoint;
+    private RuntimeEndpoint getAsyncCallback(Invocable source) {
+        if (!(source instanceof RuntimeEndpointReference))
+            return null;
+        RuntimeEndpointReference epr = (RuntimeEndpointReference)source;
+        if (!isAsyncInvocation(epr))
+            return null;
+        RuntimeEndpoint endpoint;
+        synchronized (epr) {
+            endpoint = (RuntimeEndpoint)epr.getCallbackEndpoint();
+            // If the async callback endpoint is already created, return it...
+            if (endpoint != null)
+                return endpoint;
+            // Create the endpoint for the async callback
+            endpoint = createAsyncCallbackEndpoint(epr);
+            epr.setCallbackEndpoint(endpoint);
+        } // end synchronized
+
+        // Activate the new callback endpoint
+        startEndpoint(epr.getCompositeContext(), endpoint);
+        endpoint.getInvocationChains();
+
+        return endpoint;
     } // end method setupAsyncCallback
-    
+
     /**
      * Start the callback endpoint
      * @param compositeContext - the composite context
      * @param ep - the endpoint to start
      */
-    private void startEndpoint(CompositeContext compositeContext, RuntimeEndpoint ep ) {
+    private void startEndpoint(CompositeContext compositeContext, RuntimeEndpoint ep) {
         for (PolicyProvider policyProvider : ep.getPolicyProviders()) {
             policyProvider.start();
         } // end for
@@ -388,45 +472,46 @@ public class AsyncJDKInvocationHandler extends JDKInvocationHandler {
                 public Object run() {
                     bindingProvider.start();
                     return null;
-                  }
+                }
             });
             compositeContext.getEndpointRegistry().addEndpoint(ep);
         }
     } // end method startEndpoint
-    
+
     /**
      * Create the async callback endpoint for a reference that is going to invoke an asyncInvocation service
      * @param epr - the RuntimeEndpointReference for which the callback is created
      * @return - a RuntimeEndpoint representing the callback endpoint
      */
-    private RuntimeEndpoint createAsyncCallbackEndpoint( RuntimeEndpointReference epr ) {
-    	CompositeContext compositeContext = epr.getCompositeContext();
-    	RuntimeAssemblyFactory assemblyFactory = getAssemblyFactory( compositeContext );
+    private RuntimeEndpoint createAsyncCallbackEndpoint(RuntimeEndpointReference epr) {
+        CompositeContext compositeContext = epr.getCompositeContext();
+        RuntimeAssemblyFactory assemblyFactory = getAssemblyFactory(compositeContext);
         RuntimeEndpoint endpoint = (RuntimeEndpoint)assemblyFactory.createEndpoint();
-        endpoint.bind( compositeContext );
-        
+        endpoint.bind(compositeContext);
+
         // Create a pseudo-component and pseudo-service 
         // - need to end with a chain with an invoker into the AsyncCallbackHandler class
         RuntimeComponent fakeComponent = null;
         try {
-			fakeComponent = (RuntimeComponent)epr.getComponent().clone();
-			applyImplementation( fakeComponent );
-		} catch (CloneNotSupportedException e2) {
-			// will not happen
-		} // end try
+            fakeComponent = (RuntimeComponent)epr.getComponent().clone();
+            applyImplementation(fakeComponent);
+        } catch (CloneNotSupportedException e2) {
+            // will not happen
+        } // end try
         endpoint.setComponent(fakeComponent);
-        
+
         // Create pseudo-service
         ComponentService service = assemblyFactory.createComponentService();
-    	ExtensionPointRegistry registry = compositeContext.getExtensionPointRegistry();
+        ExtensionPointRegistry registry = compositeContext.getExtensionPointRegistry();
         FactoryExtensionPoint modelFactories = registry.getExtensionPoint(FactoryExtensionPoint.class);
-        JavaInterfaceFactory javaInterfaceFactory = (JavaInterfaceFactory)modelFactories.getFactory(JavaInterfaceFactory.class);
+        JavaInterfaceFactory javaInterfaceFactory =
+            (JavaInterfaceFactory)modelFactories.getFactory(JavaInterfaceFactory.class);
         JavaInterfaceContract interfaceContract = javaInterfaceFactory.createJavaInterfaceContract();
         try {
-			interfaceContract.setInterface(javaInterfaceFactory.createJavaInterface(AsyncResponseHandler.class));
-		} catch (InvalidInterfaceException e1) {
-			// Nothing to do here - will not happen
-		} // end try
+            interfaceContract.setInterface(javaInterfaceFactory.createJavaInterface(AsyncResponseHandler.class));
+        } catch (InvalidInterfaceException e1) {
+            // Nothing to do here - will not happen
+        } // end try
         service.setInterfaceContract(interfaceContract);
         String serviceName = epr.getReference().getName() + "_asyncCallback";
         service.setName(serviceName);
@@ -435,20 +520,20 @@ public class AsyncJDKInvocationHandler extends JDKInvocationHandler {
         List<ComponentService> services = fakeComponent.getServices();
         services.clear();
         services.add(service);
-        
+
         // Create a binding
-		Binding binding = createMatchingBinding( epr.getBinding(), fakeComponent, service, registry );			
-		endpoint.setBinding(binding);
-		
-		// Need to establish policies here (binding has some...)
-		endpoint.getRequiredIntents().addAll( epr.getRequiredIntents() );
-		endpoint.getPolicySets().addAll( epr.getPolicySets() );
-		String epURI = epr.getComponent().getName() + "#service-binding(" + serviceName + "/" + serviceName + ")";
-		endpoint.setURI(epURI);
+        Binding binding = createMatchingBinding(epr.getBinding(), fakeComponent, service, registry);
+        endpoint.setBinding(binding);
+
+        // Need to establish policies here (binding has some...)
+        endpoint.getRequiredIntents().addAll(epr.getRequiredIntents());
+        endpoint.getPolicySets().addAll(epr.getPolicySets());
+        String epURI = epr.getComponent().getName() + "#service-binding(" + serviceName + "/" + serviceName + ")";
+        endpoint.setURI(epURI);
         endpoint.setUnresolved(false);
-    	return endpoint;
+        return endpoint;
     }
-    
+
     /**
      * Create a matching binding to a supplied binding
      * - the matching binding has the same binding type, but is for the supplied component and service
@@ -459,100 +544,108 @@ public class AsyncJDKInvocationHandler extends JDKInvocationHandler {
      * @return - the matching binding, or null if it could not be created
      */
     @SuppressWarnings("unchecked")
-	private Binding createMatchingBinding( Binding matchBinding, RuntimeComponent component, 
-			                               ComponentService service, ExtensionPointRegistry registry ) {
-    	// Since there is no simple way to obtain a Factory for a binding where the type is not known ahead of
-    	// time, the process followed here is to generate the <binding.xxx/> XML element from the binding type QName
-    	// and then read the XML using the processor for that XML...
-    	QName bindingName = matchBinding.getType();
-    	String bindingXML = "<ns1:" + bindingName.getLocalPart() + " xmlns:ns1='" + bindingName.getNamespaceURI() + "'/>";
-    	
-    	StAXArtifactProcessorExtensionPoint processors = registry.getExtensionPoint(StAXArtifactProcessorExtensionPoint.class);
-    	StAXArtifactProcessor<?> processor = (StAXArtifactProcessor<?>)processors.getProcessor(bindingName);
-    	
-    	FactoryExtensionPoint modelFactories = registry.getExtensionPoint(FactoryExtensionPoint.class);
-    	ValidatingXMLInputFactory inputFactory = modelFactories.getFactory(ValidatingXMLInputFactory.class);    		
-    	StreamSource source = new StreamSource( new StringReader(bindingXML) );
-    	
-    	ProcessorContext context = new ProcessorContext();
-		try {
-			XMLStreamReader reader = inputFactory.createXMLStreamReader(source);
-			reader.next();
-			Binding newBinding = (Binding) processor.read(reader, context );
-			
-			// Create a URI address for the callback based on the Component_Name/Reference_Name pattern
-			String callbackURI = "/" + component.getName() + "/" + service.getName();
-			newBinding.setURI(callbackURI);
-			
-			BuilderExtensionPoint builders = registry.getExtensionPoint(BuilderExtensionPoint.class);
-			BindingBuilder builder = builders.getBindingBuilder(newBinding.getType());
+    private Binding createMatchingBinding(Binding matchBinding,
+                                          RuntimeComponent component,
+                                          ComponentService service,
+                                          ExtensionPointRegistry registry) {
+        // Since there is no simple way to obtain a Factory for a binding where the type is not known ahead of
+        // time, the process followed here is to generate the <binding.xxx/> XML element from the binding type QName
+        // and then read the XML using the processor for that XML...
+        QName bindingName = matchBinding.getType();
+        String bindingXML =
+            "<ns1:" + bindingName.getLocalPart() + " xmlns:ns1='" + bindingName.getNamespaceURI() + "'/>";
+
+        StAXArtifactProcessorExtensionPoint processors =
+            registry.getExtensionPoint(StAXArtifactProcessorExtensionPoint.class);
+        StAXArtifactProcessor<?> processor = (StAXArtifactProcessor<?>)processors.getProcessor(bindingName);
+
+        FactoryExtensionPoint modelFactories = registry.getExtensionPoint(FactoryExtensionPoint.class);
+        ValidatingXMLInputFactory inputFactory = modelFactories.getFactory(ValidatingXMLInputFactory.class);
+        StreamSource source = new StreamSource(new StringReader(bindingXML));
+
+        ProcessorContext context = new ProcessorContext();
+        try {
+            XMLStreamReader reader = inputFactory.createXMLStreamReader(source);
+            reader.next();
+            Binding newBinding = (Binding)processor.read(reader, context);
+
+            // Create a URI address for the callback based on the Component_Name/Reference_Name pattern
+            String callbackURI = "/" + component.getName() + "/" + service.getName();
+            newBinding.setURI(callbackURI);
+
+            BuilderExtensionPoint builders = registry.getExtensionPoint(BuilderExtensionPoint.class);
+            BindingBuilder builder = builders.getBindingBuilder(newBinding.getType());
             if (builder != null) {
-            	org.apache.tuscany.sca.assembly.builder.BuilderContext builderContext = new BuilderContext(registry);
-            	builder.build(component, service, newBinding, builderContext, true);
+                org.apache.tuscany.sca.assembly.builder.BuilderContext builderContext = new BuilderContext(registry);
+                builder.build(component, service, newBinding, builderContext, true);
             } // end if
-			
-			return newBinding;
-		} catch (ContributionReadException e) {
-			e.printStackTrace();
-		} catch (XMLStreamException e) {
-			e.printStackTrace();
-		}
-    	
-    	return null;
+
+            return newBinding;
+        } catch (ContributionReadException e) {
+            e.printStackTrace();
+        } catch (XMLStreamException e) {
+            e.printStackTrace();
+        }
+
+        return null;
     } // end method createMatchingBinding
-    
+
     /**
      * Gets a RuntimeAssemblyFactory from the CompositeContext
      * @param compositeContext
      * @return the RuntimeAssemblyFactory
      */
-    private RuntimeAssemblyFactory getAssemblyFactory( CompositeContext compositeContext ) {
-    	ExtensionPointRegistry registry = compositeContext.getExtensionPointRegistry();
+    private RuntimeAssemblyFactory getAssemblyFactory(CompositeContext compositeContext) {
+        ExtensionPointRegistry registry = compositeContext.getExtensionPointRegistry();
         FactoryExtensionPoint modelFactories = registry.getExtensionPoint(FactoryExtensionPoint.class);
         return (RuntimeAssemblyFactory)modelFactories.getFactory(AssemblyFactory.class);
     } // end method RuntimeAssemblyFactory
-    
+
     /**
      * Applies an AsyncResponseHandlerImpl as the implementation of a RuntimeComponent
      * - the AsyncResponseHandlerImpl acts as both the implementation class and the implementation provider...
      * @param component - the component
      */
-    private void applyImplementation( RuntimeComponent component ) {
-    	AsyncResponseHandlerImpl<?> asyncHandler = new AsyncResponseHandlerImpl<Object>();
-    	component.setImplementation( asyncHandler );
-    	component.setImplementationProvider( asyncHandler );
+    private void applyImplementation(RuntimeComponent component) {
+        AsyncResponseHandlerImpl<?> asyncHandler = new AsyncResponseHandlerImpl<Object>();
+        component.setImplementation(asyncHandler);
+        component.setImplementationProvider(asyncHandler);
         return;
     } // end method getImplementationProvider
-    
-    private static QName ASYNC_INVOKE = new QName( Constants.SCA11_NS, "asyncInvocation" );
+
+    private static QName ASYNC_INVOKE = new QName(Constants.SCA11_NS, "asyncInvocation");
+
     /**
      * Determines if the service invocation is asynchronous
      * @param source - the EPR involved in the invocation
      * @return - true if the invocation is async
      */
-    private boolean isAsyncInvocation( Invocable source ) {
-    	if( !(source instanceof RuntimeEndpointReference) ) return false;
-		RuntimeEndpointReference epr = (RuntimeEndpointReference) source;
-		// First check is to see if the EPR itself has the asyncInvocation intent marked
-		for( Intent intent : epr.getRequiredIntents() ) {
-			if ( intent.getName().equals(ASYNC_INVOKE) ) return true;
-		} // end for
-		
-		// Second check is to see if the target service has the asyncInvocation intent marked
-		Endpoint ep = epr.getTargetEndpoint();
-		for( Intent intent : ep.getRequiredIntents() ) {
-			if ( intent.getName().equals(ASYNC_INVOKE) ) return true;
-		} // end for
-    	return false;
+    private boolean isAsyncInvocation(Invocable source) {
+        if (!(source instanceof RuntimeEndpointReference))
+            return false;
+        RuntimeEndpointReference epr = (RuntimeEndpointReference)source;
+        // First check is to see if the EPR itself has the asyncInvocation intent marked
+        for (Intent intent : epr.getRequiredIntents()) {
+            if (intent.getName().equals(ASYNC_INVOKE))
+                return true;
+        } // end for
+
+        // Second check is to see if the target service has the asyncInvocation intent marked
+        Endpoint ep = epr.getTargetEndpoint();
+        for (Intent intent : ep.getRequiredIntents()) {
+            if (intent.getName().equals(ASYNC_INVOKE))
+                return true;
+        } // end for
+        return false;
     } // end isAsyncInvocation
-    
+
     /**
      * Return the synchronous method that is the equivalent of an async method
      * @param asyncMethod - the async method
      * @return - the equivalent synchronous method
      */
     protected Method getNonAsyncMethod(Method asyncMethod) {
-        String methodName = asyncMethod.getName().substring(0, asyncMethod.getName().length()-5);
+        String methodName = asyncMethod.getName().substring(0, asyncMethod.getName().length() - 5);
         for (Method m : businessInterface.getMethods()) {
             if (methodName.equals(m.getName())) {
                 return m;
@@ -560,12 +653,12 @@ public class AsyncJDKInvocationHandler extends JDKInvocationHandler {
         }
         throw new IllegalStateException("No synchronous method matching async method " + asyncMethod.getName());
     } // end method getNonAsyncMethod
-    
+
     /**
      * Gets the classloader of the business interface
      * @return
      */
-    private ClassLoader getInterfaceClassloader( ) {
-    	return businessInterface.getClassLoader();
+    private ClassLoader getInterfaceClassloader() {
+        return businessInterface.getClassLoader();
     }
 }
diff --git a/sca-java-2.x/trunk/modules/core/src/main/java/org/apache/tuscany/sca/core/invocation/impl/JDKProxyFactory.java b/sca-java-2.x/trunk/modules/core/src/main/java/org/apache/tuscany/sca/core/invocation/impl/JDKProxyFactory.java
index 87a7f0910b..f072d66227 100644
--- a/sca-java-2.x/trunk/modules/core/src/main/java/org/apache/tuscany/sca/core/invocation/impl/JDKProxyFactory.java
+++ b/sca-java-2.x/trunk/modules/core/src/main/java/org/apache/tuscany/sca/core/invocation/impl/JDKProxyFactory.java
@@ -31,6 +31,7 @@ import javax.xml.ws.AsyncHandler;
 import javax.xml.ws.Response;
 
 import org.apache.tuscany.sca.common.java.collection.LRUCache;
+import org.apache.tuscany.sca.core.ExtensionPointRegistry;
 import org.apache.tuscany.sca.core.LifeCycleListener;
 import org.apache.tuscany.sca.core.context.ServiceReferenceExt;
 import org.apache.tuscany.sca.core.context.impl.CallbackServiceReferenceImpl;
@@ -44,17 +45,20 @@ import org.apache.tuscany.sca.runtime.RuntimeEndpoint;
 import org.oasisopen.sca.ServiceReference;
 import org.oasisopen.sca.ServiceRuntimeException;
 
-
 /**
  * the default implementation of a wire service that uses JDK dynamic proxies
  * 
  * @version $Rev$ $Date$
  */
 public class JDKProxyFactory implements ProxyFactory, LifeCycleListener {
+    protected ExtensionPointRegistry registry;
     protected InterfaceContractMapper contractMapper;
     private MessageFactory messageFactory;
 
-    public JDKProxyFactory(MessageFactory messageFactory, InterfaceContractMapper mapper) {
+    public JDKProxyFactory(ExtensionPointRegistry registry,
+                           MessageFactory messageFactory,
+                           InterfaceContractMapper mapper) {
+        this.registry = registry;
         this.contractMapper = mapper;
         this.messageFactory = messageFactory;
     }
@@ -66,17 +70,17 @@ public class JDKProxyFactory implements ProxyFactory, LifeCycleListener {
     public <T> T createProxy(final Class<T> interfaze, Invocable invocable) throws ProxyCreationException {
         if (invocable instanceof RuntimeEndpoint) {
             InvocationHandler handler;
-// TUSCANY-3659 - Always install a asynch handler regardless of whether ref is sync or async  
-//                needs tidying         
-//            if (isAsync(interfaze)) {
-                handler = new AsyncJDKInvocationHandler(messageFactory, interfaze, invocable);
-//            } else {
-//                handler = new JDKInvocationHandler(messageFactory, interfaze, invocable);
-//            }
+            // TUSCANY-3659 - Always install a asynch handler regardless of whether ref is sync or async  
+            //                needs tidying         
+            //            if (isAsync(interfaze)) {
+            handler = new AsyncJDKInvocationHandler(registry, messageFactory, interfaze, invocable);
+            //            } else {
+            //                handler = new JDKInvocationHandler(messageFactory, interfaze, invocable);
+            //            }
             // Allow privileged access to class loader. Requires RuntimePermission in security policy.
             ClassLoader cl = AccessController.doPrivileged(new PrivilegedAction<ClassLoader>() {
                 public ClassLoader run() {
-                   return interfaze.getClassLoader();
+                    return interfaze.getClassLoader();
                 }
             });
             T proxy = interfaze.cast(newProxyInstance(cl, new Class[] {interfaze}, handler));
@@ -85,35 +89,36 @@ public class JDKProxyFactory implements ProxyFactory, LifeCycleListener {
         ServiceReference<T> serviceReference = new ServiceReferenceImpl<T>(interfaze, invocable, null);
         return createProxy(serviceReference);
     }
-    
+
     public <T> T createProxy(ServiceReference<T> callableReference) throws ProxyCreationException {
         assert callableReference != null;
         final Class<T> interfaze = callableReference.getBusinessInterface();
         InvocationHandler handler;
-// TUSCANY-3659 - Always install a asynch handler regardless of whether ref is sync or async
-//                needs tidying
-//        if (isAsync(interfaze)) {
-            handler = new AsyncJDKInvocationHandler(messageFactory, callableReference);
-//        } else {
-//            handler = new JDKInvocationHandler(messageFactory, callableReference);
-//        }
+        // TUSCANY-3659 - Always install a asynch handler regardless of whether ref is sync or async
+        //                needs tidying
+        //        if (isAsync(interfaze)) {
+        handler = new AsyncJDKInvocationHandler(registry, messageFactory, callableReference);
+        //        } else {
+        //            handler = new JDKInvocationHandler(messageFactory, callableReference);
+        //        }
         // Allow privileged access to class loader. Requires RuntimePermission in security policy.
         ClassLoader cl = AccessController.doPrivileged(new PrivilegedAction<ClassLoader>() {
             public ClassLoader run() {
-               return interfaze.getClassLoader();
+                return interfaze.getClassLoader();
             }
         });
         T proxy = interfaze.cast(newProxyInstance(cl, new Class[] {interfaze}, handler));
         ((ServiceReferenceExt<T>)callableReference).setProxy(proxy);
         return proxy;
     }
-    
+
     private boolean isAsync(Class<?> interfaze) {
         for (Method method : interfaze.getMethods()) {
             if (method.getName().endsWith("Async")) {
                 if (method.getReturnType().isAssignableFrom(Future.class)) {
                     if (method.getParameterTypes().length > 0) {
-                        if (method.getParameterTypes()[method.getParameterTypes().length-1].isAssignableFrom(AsyncHandler.class)) {
+                        if (method.getParameterTypes()[method.getParameterTypes().length - 1]
+                            .isAssignableFrom(AsyncHandler.class)) {
                             return true;
                         }
                     }
@@ -143,7 +148,7 @@ public class JDKProxyFactory implements ProxyFactory, LifeCycleListener {
         InvocationHandler handler = new JDKCallbackInvocationHandler(messageFactory, callbackReference);
         ClassLoader cl = interfaze.getClassLoader();
         T proxy = interfaze.cast(newProxyInstance(cl, new Class[] {interfaze}, handler));
-        ((ServiceReferenceExt<T>) callbackReference).setProxy(proxy);
+        ((ServiceReferenceExt<T>)callbackReference).setProxy(proxy);
         return proxy;
     }
 
@@ -162,15 +167,14 @@ public class JDKProxyFactory implements ProxyFactory, LifeCycleListener {
     public boolean isProxyClass(Class<?> clazz) {
         return Proxy.isProxyClass(clazz);
     }
-    
+
     // This is a cache containing the proxy class constructor for each business interface.
     // This improves performance compared to calling Proxy.newProxyInstance()
     // every time that a proxy is needed.
     private final LRUCache<Class<?>, Constructor<?>> cache = new LRUCache<Class<?>, Constructor<?>>(512);
 
-    public Object newProxyInstance(ClassLoader classloader,
-                                          Class<?> interfaces[],
-                                          InvocationHandler invocationhandler) throws IllegalArgumentException {
+    public Object newProxyInstance(ClassLoader classloader, Class<?> interfaces[], InvocationHandler invocationhandler)
+        throws IllegalArgumentException {
         if (interfaces.length > 1) {
             // We only cache the proxy constructors with one single interface which the case in SCA where
             // one reference can have one interface
@@ -201,6 +205,6 @@ public class JDKProxyFactory implements ProxyFactory, LifeCycleListener {
     }
 
     public void stop() {
-       cache.clear(); 
-    }    
+        cache.clear();
+    }
 }
diff --git a/sca-java-2.x/trunk/modules/core/src/main/java/org/apache/tuscany/sca/core/work/impl/DefaultWorkScheduler.java b/sca-java-2.x/trunk/modules/core/src/main/java/org/apache/tuscany/sca/core/work/impl/DefaultWorkScheduler.java
index 380ca818c5..dd4abf0cd2 100644
--- a/sca-java-2.x/trunk/modules/core/src/main/java/org/apache/tuscany/sca/core/work/impl/DefaultWorkScheduler.java
+++ b/sca-java-2.x/trunk/modules/core/src/main/java/org/apache/tuscany/sca/core/work/impl/DefaultWorkScheduler.java
@@ -21,6 +21,7 @@ package org.apache.tuscany.sca.core.work.impl;
 import java.security.AccessController;
 import java.security.PrivilegedAction;
 import java.util.Map;
+import java.util.concurrent.ExecutorService;
 
 import org.apache.tuscany.sca.core.ExtensionPointRegistry;
 import org.apache.tuscany.sca.core.LifeCycleListener;
@@ -201,4 +202,9 @@ public class DefaultWorkScheduler implements WorkScheduler, LifeCycleListener {
         }
 
     }
+
+    @Override
+    public ExecutorService getExecutorService() {
+        return getWorkManager().getExecutorService();
+    }
 }
diff --git a/sca-java-2.x/trunk/modules/core/src/main/java/org/apache/tuscany/sca/core/work/impl/ThreadPoolWorkManager.java b/sca-java-2.x/trunk/modules/core/src/main/java/org/apache/tuscany/sca/core/work/impl/ThreadPoolWorkManager.java
index 1cd70c75de..7e116655e1 100644
--- a/sca-java-2.x/trunk/modules/core/src/main/java/org/apache/tuscany/sca/core/work/impl/ThreadPoolWorkManager.java
+++ b/sca-java-2.x/trunk/modules/core/src/main/java/org/apache/tuscany/sca/core/work/impl/ThreadPoolWorkManager.java
@@ -45,7 +45,7 @@ public class ThreadPoolWorkManager {
     private Map<WorkItem, WorkListener> workItems = new ConcurrentHashMap<WorkItem, WorkListener>();
 
     // Thread-pool
-    private ExecutorService executor;
+    protected ExecutorService executor;
 
     /**
      * Initializes the thread-pool.
@@ -224,5 +224,9 @@ public class ThreadPoolWorkManager {
         executor.shutdown();
     }
 
+    public ExecutorService getExecutorService() {
+        return executor;
+    }
+
 }