You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tuscany.apache.org by ed...@apache.org on 2010/07/05 13:15:25 UTC
svn commit: r960545 - in
/tuscany/sca-java-2.x/trunk/modules/core/src/main/java/org/apache/tuscany/sca/core/invocation/impl:
AsyncInvocationFutureImpl.java AsyncJDKInvocationHandler.java
Author: edwardsmj
Date: Mon Jul 5 11:15:25 2010
New Revision: 960545
URL: http://svn.apache.org/viewvc?rev=960545&view=rev
Log:
Initial version of async client endpoint for async response messages for Phase II of TUSCANY-3612
Modified:
tuscany/sca-java-2.x/trunk/modules/core/src/main/java/org/apache/tuscany/sca/core/invocation/impl/AsyncInvocationFutureImpl.java
tuscany/sca-java-2.x/trunk/modules/core/src/main/java/org/apache/tuscany/sca/core/invocation/impl/AsyncJDKInvocationHandler.java
Modified: tuscany/sca-java-2.x/trunk/modules/core/src/main/java/org/apache/tuscany/sca/core/invocation/impl/AsyncInvocationFutureImpl.java
URL: http://svn.apache.org/viewvc/tuscany/sca-java-2.x/trunk/modules/core/src/main/java/org/apache/tuscany/sca/core/invocation/impl/AsyncInvocationFutureImpl.java?rev=960545&r1=960544&r2=960545&view=diff
==============================================================================
--- tuscany/sca-java-2.x/trunk/modules/core/src/main/java/org/apache/tuscany/sca/core/invocation/impl/AsyncInvocationFutureImpl.java (original)
+++ tuscany/sca-java-2.x/trunk/modules/core/src/main/java/org/apache/tuscany/sca/core/invocation/impl/AsyncInvocationFutureImpl.java Mon Jul 5 11:15:25 2010
@@ -20,6 +20,7 @@
package org.apache.tuscany.sca.core.invocation.impl;
import java.util.Map;
+import java.util.UUID;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
@@ -30,6 +31,9 @@ import java.util.concurrent.locks.Reentr
import javax.xml.ws.Response;
+import org.apache.tuscany.sca.core.invocation.AsyncFaultWrapper;
+import org.apache.tuscany.sca.core.invocation.AsyncResponseHandler;
+
/**
* A class which provides an Implementation of a Future<V> and Response<V> for use with the JAXWS defined client
* asynchronous APIs.
@@ -41,7 +45,7 @@ import javax.xml.ws.Response;
*
* @param <V> - this is the type of the response message from the invoked service.
*/
-public class AsyncInvocationFutureImpl<V> implements Future<V>, Response<V> {
+public class AsyncInvocationFutureImpl<V> implements Future<V>, Response<V>, AsyncResponseHandler<V> {
// Lock for handling the completion of this Future
private final Lock lock = new ReentrantLock();
@@ -51,6 +55,8 @@ public class AsyncInvocationFutureImpl<V
private volatile V response = null;
private volatile Throwable fault = null;
+ private String uniqueID = UUID.randomUUID().toString();
+
protected AsyncInvocationFutureImpl() {
super();
} // end constructor
@@ -138,8 +144,10 @@ public class AsyncInvocationFutureImpl<V
* @param e - the Fault to send
* @throws IllegalStateException if either the setResponse method or the setFault method have been called previously
*/
- public void setFault(Throwable e) {
+ public void setFault(AsyncFaultWrapper w) {
+ Exception e = w.retrieveFault();
+ if( e != null ) throw new IllegalArgumentException("AsyncFaultWrapper did not return an Exception");
lock.lock();
try {
if( notSetYet() ) {
@@ -176,13 +184,18 @@ public class AsyncInvocationFutureImpl<V
} // end method setResponse
/**
+ * Gets the unique ID of this future as a String
+ */
+ public String getUniqueID() { return uniqueID; }
+
+ /**
* Indicates that setting a response value is OK - can only set the response value or fault once
* @return - true if it is OK to set the response, false otherwise
*/
private boolean notSetYet() {
return ( response == null && fault == null );
}
-
+
/**
* Returns the JAXWS context for the response
* @return - a Map containing the context
Modified: tuscany/sca-java-2.x/trunk/modules/core/src/main/java/org/apache/tuscany/sca/core/invocation/impl/AsyncJDKInvocationHandler.java
URL: http://svn.apache.org/viewvc/tuscany/sca-java-2.x/trunk/modules/core/src/main/java/org/apache/tuscany/sca/core/invocation/impl/AsyncJDKInvocationHandler.java?rev=960545&r1=960544&r2=960545&view=diff
==============================================================================
--- tuscany/sca-java-2.x/trunk/modules/core/src/main/java/org/apache/tuscany/sca/core/invocation/impl/AsyncJDKInvocationHandler.java (original)
+++ tuscany/sca-java-2.x/trunk/modules/core/src/main/java/org/apache/tuscany/sca/core/invocation/impl/AsyncJDKInvocationHandler.java Mon Jul 5 11:15:25 2010
@@ -19,17 +19,64 @@
package org.apache.tuscany.sca.core.invocation.impl;
+import java.io.StringReader;
import java.lang.reflect.Method;
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.Future;
+import javax.xml.namespace.QName;
+import javax.xml.stream.XMLInputFactory;
+import javax.xml.stream.XMLStreamException;
+import javax.xml.stream.XMLStreamReader;
+import javax.xml.transform.stream.StreamSource;
import javax.xml.ws.AsyncHandler;
import javax.xml.ws.Response;
+import org.apache.tuscany.sca.assembly.AssemblyFactory;
+import org.apache.tuscany.sca.assembly.Binding;
+import org.apache.tuscany.sca.assembly.ComponentService;
+import org.apache.tuscany.sca.assembly.Endpoint;
+import org.apache.tuscany.sca.assembly.Implementation;
+import org.apache.tuscany.sca.assembly.builder.BindingBuilder;
+import org.apache.tuscany.sca.assembly.builder.BuilderContext;
+import org.apache.tuscany.sca.assembly.builder.BuilderExtensionPoint;
+import org.apache.tuscany.sca.assembly.xml.Constants;
+import org.apache.tuscany.sca.context.CompositeContext;
+import org.apache.tuscany.sca.contribution.processor.ContributionReadException;
+import org.apache.tuscany.sca.contribution.processor.ProcessorContext;
+import org.apache.tuscany.sca.contribution.processor.StAXArtifactProcessor;
+import org.apache.tuscany.sca.contribution.processor.StAXArtifactProcessorExtensionPoint;
+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.assembly.RuntimeAssemblyFactory;
+import org.apache.tuscany.sca.core.assembly.impl.RuntimeEndpointReferenceImpl;
+import org.apache.tuscany.sca.core.invocation.AsyncFaultWrapper;
+import org.apache.tuscany.sca.core.invocation.AsyncResponseHandler;
+import org.apache.tuscany.sca.interfacedef.InterfaceContract;
+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.invocation.InvocationChain;
import org.apache.tuscany.sca.invocation.MessageFactory;
+import org.apache.tuscany.sca.policy.Intent;
+import org.apache.tuscany.sca.provider.ImplementationProvider;
+import org.apache.tuscany.sca.provider.ImplementationProviderFactory;
+import org.apache.tuscany.sca.provider.PolicyProvider;
+import org.apache.tuscany.sca.provider.RuntimeProvider;
+import org.apache.tuscany.sca.provider.ServiceBindingProvider;
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.oasisopen.sca.ComponentContext;
import org.oasisopen.sca.ServiceReference;
+import org.oasisopen.sca.ServiceRuntimeException;
+import org.oasisopen.sca.annotation.AsyncInvocation;
/**
* An InvocationHandler which deals with JAXWS-defined asynchronous client Java API method calls
@@ -107,24 +154,25 @@ public class AsyncJDKInvocationHandler e
* @return - the Response<?> object that is returned to the client application, typed by the
* type of the response
*/
- protected Response doInvokeAsyncPoll(Object proxy, Method asyncMethod, Object[] args) {
+ @SuppressWarnings("unchecked")
+ protected Response doInvokeAsyncPoll(Object proxy, Method asyncMethod, Object[] args) {
Object response;
- boolean isException;
Class<?> returnType = getNonAsyncMethod(asyncMethod).getReturnType();
// Allocate the Future<?> / Response<?> object - note: Response<?> is a subclass of Future<?>
AsyncInvocationFutureImpl future = AsyncInvocationFutureImpl.newInstance( returnType );
try {
- response = super.invoke(proxy, getNonAsyncMethod(asyncMethod), args);
- isException = false;
+ response = invokeAsync(proxy, getNonAsyncMethod(asyncMethod), args, future);
future.setResponse(response);
- } catch (Throwable e) {
- response = e;
- isException = true;
- future.setFault(e);
- }
+ } 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) );
+ } // end try
return future;
//return new AsyncResponse(response, isException);
- }
+ } // end method doInvokeAsyncPoll
/**
* Invoke an async callback method
@@ -134,14 +182,273 @@ public class AsyncJDKInvocationHandler e
* @return - the Future<?> object that is returned to the client application, typed by the type of
* the response
*/
- private Object doInvokeAsyncCallback(Object proxy, Method asyncMethod, Object[] args) {
+ @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));
handler.handleResponse(response);
return response;
- }
+ } // end method doInvokeAsyncCallback
+
+ /**
+ * Invoke the target method on
+ * @param proxy
+ * @param method - the method to invoke
+ * @param args - arguments for the call
+ * @param future - Future for handling the response
+ * @return - returns the response from the invocation
+ * @throws Throwable - if an exception is thrown during the invocation
+ */
+ @SuppressWarnings("unchecked")
+ private Object invokeAsync(Object proxy, Method method, Object[] args, AsyncInvocationFutureImpl future) throws Throwable {
+ if (Object.class == method.getDeclaringClass()) {
+ return invokeObjectMethod(method, args);
+ }
+ if (source == null) {
+ throw new ServiceRuntimeException("No runtime source is available");
+ }
+
+ if (source instanceof RuntimeEndpointReference) {
+ RuntimeEndpointReference epr = (RuntimeEndpointReference)source;
+ if (epr.isOutOfDate()) {
+ epr.rebuild();
+ chains.clear();
+ }
+ }
+
+ InvocationChain chain = getInvocationChain(method, source);
+
+ if (chain == null) {
+ throw new IllegalArgumentException("No matching operation is found: " + method);
+ }
+
+ RuntimeEndpoint theEndpoint = getAsyncCallback( source );
+ attachFuture( theEndpoint, future );
+
+ // send the invocation down the source
+ Object result = super.invoke(chain, args, source);
+ return result;
+ } // end method invokeAsync
+
+ /**
+ * 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);
+ } // 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( endpoint != null ) return endpoint;
+ // Create the endpoint for the async callback
+ endpoint = createAsyncCallbackEndpoint( epr );
+ epr.setCallbackEndpoint(endpoint);
+ }
+
+ // 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 ) {
+ for (PolicyProvider policyProvider : ep.getPolicyProviders()) {
+ policyProvider.start();
+ } // end for
+
+ final ServiceBindingProvider bindingProvider = ep.getBindingProvider();
+ if (bindingProvider != null) {
+ // Allow bindings to add shutdown hooks. Requires RuntimePermission shutdownHooks in policy.
+ AccessController.doPrivileged(new PrivilegedAction<Object>() {
+ 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 );
+ RuntimeEndpoint endpoint = (RuntimeEndpoint)assemblyFactory.createEndpoint();
+ 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
+ endpoint.setComponent(fakeComponent);
+
+ // Create pseudo-service
+ ComponentService service = assemblyFactory.createComponentService();
+ ExtensionPointRegistry registry = compositeContext.getExtensionPointRegistry();
+ FactoryExtensionPoint modelFactories = registry.getExtensionPoint(FactoryExtensionPoint.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
+ service.setInterfaceContract(interfaceContract);
+ String serviceName = epr.getReference().getName() + "_asyncCallback";
+ service.setName(serviceName);
+ endpoint.setService(service);
+ // Set pseudo-service onto the pseudo-component
+ List<ComponentService> services = fakeComponent.getServices();
+ services.clear();
+ services.add(service);
+
+ Binding eprBinding = epr.getBinding();
+ try {
+ Binding binding = (Binding)eprBinding.clone();
+ // Create a binding
+ binding = createMatchingBinding( eprBinding, fakeComponent, service, registry );
+
+ // Create a URI address for the callback based on the Component_Name/Reference_Name pattern
+ //String callbackURI = "/" + epr.getComponent().getName() + "/" + serviceName;
+ //binding.setURI(callbackURI);
+ endpoint.setBinding(binding);
+ } catch (CloneNotSupportedException e) {
+ // will not happen
+ } // end try
+
+ // 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;
+ }
+
+ /**
+ * Create a matching binding to a supplied binding
+ * - the matching binding has the same binding type, but is for the supplied component and service
+ * @param matchBinding - the binding to match
+ * @param component - the component
+ * @param service - the service
+ * @param registry - registry for extensions
+ * @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());
+ if (builder != null) {
+ 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;
+ } // end method createMatchingBinding
+
+ /**
+ * Gets a RuntimeAssemblyFactory from the CompositeContext
+ * @param compositeContext
+ * @return the RuntimeAssemblyFactory
+ */
+ private RuntimeAssemblyFactory getAssemblyFactory( CompositeContext compositeContext ) {
+ ExtensionPointRegistry registry = compositeContext.getExtensionPointRegistry();
+ FactoryExtensionPoint modelFactories = registry.getExtensionPoint(FactoryExtensionPoint.class);
+ return (RuntimeAssemblyFactory)modelFactories.getFactory(AssemblyFactory.class);
+ } // end method RuntimeAssemblyFactory
+
+ 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" );
+ /**
+ * 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( RuntimeEndpointReference source ) {
+ 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