diff options
author | rfeng <rfeng@13f79535-47bb-0310-9956-ffa450edef68> | 2010-09-17 20:37:49 +0000 |
---|---|---|
committer | rfeng <rfeng@13f79535-47bb-0310-9956-ffa450edef68> | 2010-09-17 20:37:49 +0000 |
commit | d0be875c6aacb02435860e82db3feebea99132ad (patch) | |
tree | 02c18b48dc59828e50a8815fe7280c68a70b77e4 | |
parent | 1bce2af6ab834ea0985d8ef855118c0a37dc794e (diff) |
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
Diffstat (limited to '')
13 files changed, 595 insertions, 384 deletions
diff --git a/sca-java-2.x/trunk/modules/binding-jms-runtime/src/main/java/org/apache/tuscany/sca/binding/jms/provider/RRBJMSBindingInvoker.java b/sca-java-2.x/trunk/modules/binding-jms-runtime/src/main/java/org/apache/tuscany/sca/binding/jms/provider/RRBJMSBindingInvoker.java index b578d7e2a5..defd8f222a 100644 --- a/sca-java-2.x/trunk/modules/binding-jms-runtime/src/main/java/org/apache/tuscany/sca/binding/jms/provider/RRBJMSBindingInvoker.java +++ b/sca-java-2.x/trunk/modules/binding-jms-runtime/src/main/java/org/apache/tuscany/sca/binding/jms/provider/RRBJMSBindingInvoker.java @@ -136,6 +136,11 @@ public class RRBJMSBindingInvoker implements Invoker { qCreateMode = jmsBinding.getDestinationCreate(); } + // FIXME: [rfeng] A hack to remove jms:jndi: prefix + if (queueName.startsWith("jms:jndi:")) { + queueName = queueName.substring("jms:jndi:".length()); + } + Destination dest = jmsResourceFactory.lookupDestination(queueName); if (qCreateMode.equals(JMSBindingConstants.CREATE_ALWAYS)) { diff --git a/sca-java-2.x/trunk/modules/core-spi/src/main/java/org/apache/tuscany/sca/work/WorkScheduler.java b/sca-java-2.x/trunk/modules/core-spi/src/main/java/org/apache/tuscany/sca/work/WorkScheduler.java index 2c5fb556bc..f8a1ef4d23 100644 --- a/sca-java-2.x/trunk/modules/core-spi/src/main/java/org/apache/tuscany/sca/work/WorkScheduler.java +++ b/sca-java-2.x/trunk/modules/core-spi/src/main/java/org/apache/tuscany/sca/work/WorkScheduler.java @@ -18,6 +18,8 @@ */ package org.apache.tuscany.sca.work; +import java.util.concurrent.ExecutorService; + /** * Defines the contract for scheduling asynchronous units of work. * @@ -50,5 +52,11 @@ public interface WorkScheduler { * @param work The unit of work that needs to be asynchronously executed. */ <T extends Runnable>void scheduleWork(T work); + + /** + * Get the underlying getExecutorService + * @return + */ + ExecutorService getExecutorService(); } 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; + } + } diff --git a/sca-java-2.x/trunk/modules/host-jetty/src/test/java/org/apache/tuscany/sca/http/jetty/JettyServerTestCase.java b/sca-java-2.x/trunk/modules/host-jetty/src/test/java/org/apache/tuscany/sca/http/jetty/JettyServerTestCase.java index a3199cf4d1..a83246174c 100644 --- a/sca-java-2.x/trunk/modules/host-jetty/src/test/java/org/apache/tuscany/sca/http/jetty/JettyServerTestCase.java +++ b/sca-java-2.x/trunk/modules/host-jetty/src/test/java/org/apache/tuscany/sca/http/jetty/JettyServerTestCase.java @@ -27,6 +27,8 @@ import java.net.ConnectException; import java.net.InetAddress; import java.net.Socket; import java.net.URL; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; import javax.net.ssl.HostnameVerifier; import javax.net.ssl.HttpsURLConnection; @@ -70,17 +72,18 @@ public class JettyServerTestCase extends TestCase { private static final int HTTP_PORT = 8085; private WorkScheduler workScheduler = new WorkScheduler() { - + private ExecutorService executorService = Executors.newCachedThreadPool(); + public <T extends Runnable> void scheduleWork(T work) { - Thread thread = new Thread(work); - thread.start(); + executorService.submit(work); } public <T extends Runnable> void scheduleWork(T work, NotificationListener<T> listener) { scheduleWork(work); } - public void destroy() { + public ExecutorService getExecutorService() { + return executorService; } }; diff --git a/sca-java-2.x/trunk/modules/interface-java-jaxws/pom.xml b/sca-java-2.x/trunk/modules/interface-java-jaxws/pom.xml index a1e93fd077..a628d825da 100644 --- a/sca-java-2.x/trunk/modules/interface-java-jaxws/pom.xml +++ b/sca-java-2.x/trunk/modules/interface-java-jaxws/pom.xml @@ -213,11 +213,35 @@ <wsdlFiles> <wsdlFile>StockExceptionTest.wsdl</wsdlFile> </wsdlFiles> + <staleFile>${project.build.directory}/wsimport-sync.stale</staleFile> <sourceDestDir>${project.build.directory}/jaxws-source</sourceDestDir> <verbose>false</verbose> <xnocompile>true</xnocompile> </configuration> </execution> + + <execution> + <id>wsimport-async</id> + <phase>generate-test-sources</phase> + <goals> + <goal>wsimport</goal> + </goals> + <configuration> + <packageName>com.example.stock.async</packageName> + <wsdlDirectory>${basedir}/src/test/resources/wsdl</wsdlDirectory> + <wsdlFiles> + <wsdlFile>StockExceptionTest.wsdl</wsdlFile> + </wsdlFiles> + <bindingDirectory>${basedir}/src/test/resources/wsdl</bindingDirectory> + <bindingFiles> + <bindingFile>bindings.xml</bindingFile> + </bindingFiles> + <staleFile>${project.build.directory}/wsimport-async.stale</staleFile> + <sourceDestDir>${project.build.directory}/jaxws-source</sourceDestDir> + <verbose>false</verbose> + <xnocompile>true</xnocompile> + </configuration> + </execution> <!-- <execution> diff --git a/sca-java-2.x/trunk/modules/interface-java-jaxws/src/main/java/org/apache/tuscany/sca/interfacedef/java/jaxws/JAXWSAsyncInterfaceProcessor.java b/sca-java-2.x/trunk/modules/interface-java-jaxws/src/main/java/org/apache/tuscany/sca/interfacedef/java/jaxws/JAXWSAsyncInterfaceProcessor.java index 155c89c7f2..a866c5b8e1 100644 --- a/sca-java-2.x/trunk/modules/interface-java-jaxws/src/main/java/org/apache/tuscany/sca/interfacedef/java/jaxws/JAXWSAsyncInterfaceProcessor.java +++ b/sca-java-2.x/trunk/modules/interface-java-jaxws/src/main/java/org/apache/tuscany/sca/interfacedef/java/jaxws/JAXWSAsyncInterfaceProcessor.java @@ -20,48 +20,54 @@ package org.apache.tuscany.sca.interfacedef.java.jaxws; import java.lang.reflect.ParameterizedType; +import java.lang.reflect.Type; import java.util.ArrayList; import java.util.List; +import java.util.concurrent.Future; + +import javax.xml.ws.AsyncHandler; +import javax.xml.ws.Response; import org.apache.tuscany.sca.core.ExtensionPointRegistry; import org.apache.tuscany.sca.interfacedef.DataType; import org.apache.tuscany.sca.interfacedef.InvalidInterfaceException; import org.apache.tuscany.sca.interfacedef.Operation; import org.apache.tuscany.sca.interfacedef.java.JavaInterface; +import org.apache.tuscany.sca.interfacedef.java.JavaOperation; import org.apache.tuscany.sca.interfacedef.java.introspect.JavaInterfaceVisitor; public class JAXWSAsyncInterfaceProcessor implements JavaInterfaceVisitor { private static String ASYNC = "Async"; public JAXWSAsyncInterfaceProcessor(ExtensionPointRegistry registry) { - } - + public void visitInterface(JavaInterface javaInterface) throws InvalidInterfaceException { List<Operation> validOperations = new ArrayList<Operation>(); List<Operation> asyncOperations = new ArrayList<Operation>(); - + validOperations.addAll(javaInterface.getOperations()); - for(Operation o : javaInterface.getOperations()) { - if (! o.getName().endsWith(ASYNC)) { - Operation op = o; - - for(Operation asyncOp : getAsyncOperations(javaInterface.getOperations(), o.getName()) ) { - if (isJAXWSAsyncPoolingOperation(op, asyncOp) || - isJAXWSAsyncCallbackOperation(op, asyncOp)) { + for (Operation o : javaInterface.getOperations()) { + if (!o.getName().endsWith(ASYNC)) { + JavaOperation op = (JavaOperation)o; + if (op.getJavaMethod().getName().endsWith(ASYNC)) { + continue; + } + for (Operation asyncOp : getAsyncOperations(javaInterface.getOperations(), op)) { + if (isJAXWSAsyncPoolingOperation(op, asyncOp) || isJAXWSAsyncCallbackOperation(op, asyncOp)) { validOperations.remove(asyncOp); asyncOperations.add(asyncOp); - } + } } - } + } } - + javaInterface.getOperations().clear(); javaInterface.getOperations().addAll(validOperations); - + javaInterface.getAttributes().put("JAXWS-ASYNC-OPERATIONS", asyncOperations); } - + /** * The additional client-side asynchronous polling and callback methods defined by JAX-WS are recognized in a Java interface as follows: * For each method M in the interface, if another method P in the interface has @@ -75,53 +81,59 @@ public class JAXWSAsyncInterfaceProcessor implements JavaInterfaceVisitor { * @return */ private static boolean isJAXWSAsyncPoolingOperation(Operation operation, Operation asyncOperation) { - //a method name that is M's method name with the characters "Async" appended - if (operation.getName().endsWith(ASYNC)) { - return false; - } - if (! asyncOperation.getName().endsWith(ASYNC)) { + if (asyncOperation.getOutputType() == null || Response.class != asyncOperation.getOutputType().getPhysical()) { + // The return type is not Response<T> return false; } - - if(! asyncOperation.getName().equals(operation.getName() + ASYNC)) { - return false; - } - + //the same parameter signature as M List<DataType> operationInputType = operation.getInputType().getLogical(); List<DataType> asyncOperationInputType = asyncOperation.getInputType().getLogical(); int size = operationInputType.size(); + if (asyncOperationInputType.size() != size) { + return false; + } for (int i = 0; i < size; i++) { if (!isCompatible(operationInputType.get(i), asyncOperationInputType.get(i))) { return false; } } - + //a return type of Response<R> where R is the return type of M DataType<?> operationOutputType = operation.getOutputType(); DataType<?> asyncOperationOutputType = asyncOperation.getOutputType(); - + if (operationOutputType != null && asyncOperationOutputType != null) { - ParameterizedType asyncReturnType = (ParameterizedType) asyncOperationOutputType.getGenericType(); - Class<?> asyncReturnTypeClass = (Class<?>)asyncReturnType.getRawType(); - if(asyncReturnTypeClass.getName().equals("javax.xml.ws.Response")) { + Class<?> asyncReturnTypeClass = (Class<?>)asyncOperationOutputType.getPhysical(); + if (asyncReturnTypeClass == Response.class) { //now check the actual type of the Response<R> with R - Class<?> returnType = operationOutputType.getPhysical(); - Class<?> asyncActualReturnTypeClass = (Class<?>) asyncReturnType.getActualTypeArguments()[0]; - - if(returnType == asyncActualReturnTypeClass || - returnType.isPrimitive() && primitiveAssignable(returnType,asyncActualReturnTypeClass)) { - //valid - } else { - return false; + Class<?> returnType = operationOutputType.getPhysical(); + Class<?> asyncActualReturnTypeClass = Object.class; + if (asyncOperationOutputType.getGenericType() instanceof ParameterizedType) { + ParameterizedType asyncReturnType = (ParameterizedType)asyncOperationOutputType.getGenericType(); + asyncActualReturnTypeClass = (Class<?>)asyncReturnType.getActualTypeArguments()[0]; + } + + if (operation.getWrapper() != null) { + // The return type could be the wrapper type per JAX-WS spec + Class<?> wrapperClass = operation.getWrapper().getOutputWrapperClass(); + if (wrapperClass == asyncActualReturnTypeClass) { + return true; + } + } + if (returnType == asyncActualReturnTypeClass || returnType.isPrimitive() + && primitiveAssignable(returnType, asyncActualReturnTypeClass)) { + return true; + } else { + return false; } } } - + return true; } - + /** * For each method M in the interface, if another method C in the interface has * a) a method name that is M's method name with the characters "Async" appended, and @@ -136,71 +148,57 @@ public class JAXWSAsyncInterfaceProcessor implements JavaInterfaceVisitor { * @return */ private static boolean isJAXWSAsyncCallbackOperation(Operation operation, Operation asyncOperation) { - //a method name that is M's method name with the characters "Async" appended - if (operation.getName().endsWith(ASYNC)) { - return false; - } - if (! asyncOperation.getName().endsWith(ASYNC)) { - return false; - } - - if(! asyncOperation.getName().equals(operation.getName() + ASYNC)) { + if (asyncOperation.getOutputType() == null || Future.class != asyncOperation.getOutputType().getPhysical()) { + // The return type is not Future<?> return false; } - + //a parameter signature that is M's parameter signature //with an additional final parameter of type AsyncHandler<R> where R is the return type of M, and List<DataType> operationInputType = operation.getInputType().getLogical(); List<DataType> asyncOperationInputType = asyncOperation.getInputType().getLogical(); int size = operationInputType.size(); + if (asyncOperationInputType.size() != size + 1) { + return false; + } for (int i = 0; i < size; i++) { if (!isCompatible(operationInputType.get(i), asyncOperationInputType.get(i))) { return false; } } - - if(asyncOperationInputType.size() == size + 1) { - ParameterizedType asyncLastParameterType = (ParameterizedType) asyncOperationInputType.get(size + 1).getGenericType(); - Class<?> asyncLastParameterTypeClass = (Class<?>)asyncLastParameterType.getRawType(); - if(asyncLastParameterTypeClass.getName().equals("javax.xml.ws.AsyncHandler")) { - //now check the actual type of the AsyncHandler<R> with R - Class<?> returnType = operation.getOutputType().getPhysical(); - Class<?> asyncActualLastParameterTypeClass = (Class<?>) asyncLastParameterType.getActualTypeArguments()[0]; - - if(returnType == asyncActualLastParameterTypeClass || - returnType.isPrimitive() && primitiveAssignable(returnType,asyncActualLastParameterTypeClass)) { - //valid - } else { - return false; - } - } - } - - //a return type of Response<R> where R is the return type of M - DataType<?> operationOutputType = operation.getOutputType(); - DataType<?> asyncOperationOutputType = asyncOperation.getOutputType(); - - if (operationOutputType != null && asyncOperationOutputType != null) { - ParameterizedType asyncReturnType = (ParameterizedType) asyncOperationOutputType.getGenericType(); - Class<?> asyncReturnTypeClass = (Class<?>)asyncReturnType.getRawType(); - if(asyncReturnTypeClass.getName().equals("javax.xml.ws.Response")) { - //now check the actual type of the Response<R> with R - Class<?> returnType = operationOutputType.getPhysical(); - Class<?> asyncActualReturnTypeClass = (Class<?>) asyncReturnType.getActualTypeArguments()[0]; - - if(returnType == asyncActualReturnTypeClass || - returnType.isPrimitive() && primitiveAssignable(returnType,asyncActualReturnTypeClass)) { - //valid - } else { - return false; + + Type genericParamType = asyncOperationInputType.get(size).getGenericType(); + + Class<?> asyncLastParameterTypeClass = asyncOperationInputType.get(size).getPhysical(); + if (asyncLastParameterTypeClass == AsyncHandler.class) { + //now check the actual type of the AsyncHandler<R> with R + Class<?> returnType = operation.getOutputType().getPhysical(); + Class<?> asyncActualLastParameterTypeClass = Object.class; + if (genericParamType instanceof ParameterizedType) { + ParameterizedType asyncLastParameterType = (ParameterizedType)genericParamType; + asyncActualLastParameterTypeClass = (Class<?>)asyncLastParameterType.getActualTypeArguments()[0]; + } + + if (operation.getWrapper() != null) { + // The return type could be the wrapper type per JAX-WS spec + Class<?> wrapperClass = operation.getWrapper().getOutputWrapperClass(); + if (wrapperClass == asyncActualLastParameterTypeClass) { + return true; } } + + if (returnType == asyncActualLastParameterTypeClass || returnType.isPrimitive() + && primitiveAssignable(returnType, asyncActualLastParameterTypeClass)) { + return true; + } else { + return false; + } } - + return true; - } - + } + /** * Get operation by name * @@ -208,18 +206,30 @@ public class JAXWSAsyncInterfaceProcessor implements JavaInterfaceVisitor { * @param operationName * @return */ - private static List<Operation> getAsyncOperations(List<Operation> operations, String operationName) { + private static List<Operation> getAsyncOperations(List<Operation> operations, JavaOperation op) { List<Operation> returnOperations = new ArrayList<Operation>(); - for(Operation o : operations) { - if(o.getName().equals(operationName + ASYNC)) { + for (Operation o : operations) { + if (o == op) { + continue; + } + String operationName = op.getName(); + if (o.getName().equals(operationName)) { + // Async operations have the same name when @WebMethod is present + /* + JavaOperation jop = (JavaOperation)o; + if (op.getJavaMethod().getName().equals(jop.getJavaMethod().getName() + ASYNC)) { + returnOperations.add(o); + } + */ + returnOperations.add(o); + } else if (o.getName().equals(operationName + ASYNC)) { returnOperations.add(o); } } - + return returnOperations; } - /** * Check if two operation parameters are compatible diff --git a/sca-java-2.x/trunk/modules/interface-java-jaxws/src/main/resources/META-INF/services/org.apache.tuscany.sca.interfacedef.java.introspect.JavaInterfaceVisitor b/sca-java-2.x/trunk/modules/interface-java-jaxws/src/main/resources/META-INF/services/org.apache.tuscany.sca.interfacedef.java.introspect.JavaInterfaceVisitor index 86255f4fef..05b1ccea9b 100644 --- a/sca-java-2.x/trunk/modules/interface-java-jaxws/src/main/resources/META-INF/services/org.apache.tuscany.sca.interfacedef.java.introspect.JavaInterfaceVisitor +++ b/sca-java-2.x/trunk/modules/interface-java-jaxws/src/main/resources/META-INF/services/org.apache.tuscany.sca.interfacedef.java.introspect.JavaInterfaceVisitor @@ -13,7 +13,11 @@ # "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY # KIND, either express or implied. See the License for the # specific language governing permissions and limitations -# under the License. +# under the License. +# +# first one, set remotable flag org.apache.tuscany.sca.interfacedef.java.jaxws.WebServiceInterfaceProcessor;ranking=400 +# second one, introspect the JAXWS annotations org.apache.tuscany.sca.interfacedef.java.jaxws.JAXWSJavaInterfaceProcessor;ranking=100 -org.apache.tuscany.sca.interfacedef.java.jaxws.JAXWSAsyncInterfaceProcessor;ranking=900
+# third one, introspect JAXWS async methods +org.apache.tuscany.sca.interfacedef.java.jaxws.JAXWSAsyncInterfaceProcessor;ranking=50
diff --git a/sca-java-2.x/trunk/modules/interface-java-jaxws/src/test/java/org/apache/tuscany/sca/interfacedef/java/jaxws/JAXWSAsyncInterfaceProcessorTestCase.java b/sca-java-2.x/trunk/modules/interface-java-jaxws/src/test/java/org/apache/tuscany/sca/interfacedef/java/jaxws/JAXWSAsyncInterfaceProcessorTestCase.java index ec73ad4218..c53521db00 100644 --- a/sca-java-2.x/trunk/modules/interface-java-jaxws/src/test/java/org/apache/tuscany/sca/interfacedef/java/jaxws/JAXWSAsyncInterfaceProcessorTestCase.java +++ b/sca-java-2.x/trunk/modules/interface-java-jaxws/src/test/java/org/apache/tuscany/sca/interfacedef/java/jaxws/JAXWSAsyncInterfaceProcessorTestCase.java @@ -33,6 +33,8 @@ import org.apache.tuscany.sca.interfacedef.java.JavaInterface; import org.junit.Before; import org.junit.Test; +import com.example.stock.async.StockExceptionTest; + public class JAXWSAsyncInterfaceProcessorTestCase { private ExtensionPointRegistry registry; @@ -64,4 +66,24 @@ public class JAXWSAsyncInterfaceProcessorTestCase { } } + + @Test + public final void testProcessorGenerated() throws Exception { + DefaultJavaInterfaceFactory iFactory = new DefaultJavaInterfaceFactory(registry); + JavaInterface contract = iFactory.createJavaInterface(StockExceptionTest.class); + + assertTrue(contract.isRemotable()); + + Assert.assertEquals(1,contract.getOperations().size()); + + List<Operation> asyncOperations = (List<Operation>) contract.getAttributes().get("JAXWS-ASYNC-OPERATIONS"); + Assert.assertEquals(2,asyncOperations.size()); + + //list operation + System.out.println(">>> Filtered Operations"); + for(Operation o : contract.getOperations()) { + System.out.println(">>>>>>" + o); + } + + } } diff --git a/sca-java-2.x/trunk/modules/interface-java-jaxws/src/test/resources/wsdl/bindings.xml b/sca-java-2.x/trunk/modules/interface-java-jaxws/src/test/resources/wsdl/bindings.xml new file mode 100644 index 0000000000..078abb5ac1 --- /dev/null +++ b/sca-java-2.x/trunk/modules/interface-java-jaxws/src/test/resources/wsdl/bindings.xml @@ -0,0 +1,28 @@ +<?xml version="1.0" encoding="UTF-8"?> +<!-- + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. +--> +<bindings + xmlns:xsd="http://www.w3.org/2001/XMLSchema" + xmlns:wsdl="http://schemas.xmlsoap.org/wsdl/" + wsdlLocation="StockExceptionTest.wsdl" + xmlns="http://java.sun.com/xml/ns/jaxws"> + <bindings node="wsdl:definitions"> + <enableAsyncMapping>true</enableAsyncMapping> + </bindings> +</bindings> |