1 files changed, 351 insertions, 0 deletions
diff --git a/sca-java-2.x/tags/2.0.1-RC1/modules/core/src/main/java/org/apache/tuscany/sca/core/invocation/AsyncResponseInvoker.java b/sca-java-2.x/tags/2.0.1-RC1/modules/core/src/main/java/org/apache/tuscany/sca/core/invocation/AsyncResponseInvoker.java
new file mode 100644
index 0000000000..1ebc9c633a
--- /dev/null
+++ b/sca-java-2.x/tags/2.0.1-RC1/modules/core/src/main/java/org/apache/tuscany/sca/core/invocation/AsyncResponseInvoker.java
@@ -0,0 +1,351 @@
+/*
+ * 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.    
+ */
+
+package org.apache.tuscany.sca.core.invocation;
+
+import java.io.IOException;
+import java.io.ObjectInputStream;
+import java.io.Serializable;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.tuscany.sca.assembly.Endpoint;
+import org.apache.tuscany.sca.assembly.EndpointReference;
+import org.apache.tuscany.sca.context.CompositeContext;
+import org.apache.tuscany.sca.core.ExtensionPointRegistry;
+import org.apache.tuscany.sca.core.ExtensionPointRegistryLocator;
+import org.apache.tuscany.sca.core.FactoryExtensionPoint;
+import org.apache.tuscany.sca.interfacedef.Operation;
+import org.apache.tuscany.sca.invocation.InvokerAsyncResponse;
+import org.apache.tuscany.sca.invocation.Message;
+import org.apache.tuscany.sca.invocation.MessageFactory;
+import org.apache.tuscany.sca.provider.EndpointAsyncProvider;
+import org.apache.tuscany.sca.runtime.DomainRegistryFactory;
+import org.apache.tuscany.sca.runtime.DomainRegistry;
+import org.apache.tuscany.sca.runtime.ExtensibleDomainRegistryFactory;
+import org.apache.tuscany.sca.runtime.RuntimeEndpoint;
+import org.apache.tuscany.sca.runtime.RuntimeEndpointReference;
+
+/**
+ * A class that wraps the mechanics for sending async responses
+ * and hides the decision about whether the response will be processed
+ * natively or non-natively
+ * 
+ * This class is generic, based on the type of targetAddress information required by
+ * the Binding that creates it
+ */
+public class AsyncResponseInvoker<T> implements InvokerAsyncResponse, Serializable {
+    
+    /**
+	 * 
+	 */
+	private static final long serialVersionUID = -7992598227671386588L;
+	
+	private transient RuntimeEndpoint requestEndpoint;
+    private transient RuntimeEndpointReference responseEndpointReference; 
+    private T responseTargetAddress;
+    private String relatesToMsgID;
+    private String operationName;
+    private transient MessageFactory messageFactory;
+    private String bindingType = "";
+    private boolean isNativeAsync;
+    
+    private String endpointURI;
+    private String endpointReferenceURI;
+    private String domainURI;
+
+	private transient DomainRegistry domainRegistry;
+	private transient ExtensionPointRegistry registry;
+    
+    public AsyncResponseInvoker(RuntimeEndpoint requestEndpoint,
+			RuntimeEndpointReference responseEndpointReference,
+			T responseTargetAddress, String relatesToMsgID, 
+			String operationName, MessageFactory messageFactory) {
+		super();
+		this.requestEndpoint = requestEndpoint;
+		this.responseEndpointReference = responseEndpointReference;
+		this.responseTargetAddress = responseTargetAddress;
+		this.relatesToMsgID = relatesToMsgID;
+		this.operationName = operationName;
+		this.messageFactory = messageFactory;
+		
+		CompositeContext context = null;
+		if(requestEndpoint != null ) {
+			endpointURI = requestEndpoint.getURI();
+			context = requestEndpoint.getCompositeContext(); 
+		} // end if
+		if(responseEndpointReference != null ) {
+			endpointReferenceURI = responseEndpointReference.getURI();
+			context = responseEndpointReference.getCompositeContext();
+		}
+		
+		if( context != null ) {
+			domainURI = context.getDomainURI();
+			registry = context.getExtensionPointRegistry();
+		} // end if
+		
+        if ((requestEndpoint.getBindingProvider() instanceof EndpointAsyncProvider) &&
+                (((EndpointAsyncProvider)requestEndpoint.getBindingProvider()).supportsNativeAsync())){
+        	isNativeAsync = true;
+        } else {
+        	isNativeAsync = false;
+        } // end if
+	} // end constructor
+
+    /** 
+     * If you have a Tuscany message you can call this
+     */
+    public void invokeAsyncResponse(Message responseMessage) {
+    	responseMessage.getHeaders().put(Constants.ASYNC_RESPONSE_INVOKER, this);
+    	responseMessage.getHeaders().put(Constants.RELATES_TO, relatesToMsgID);
+    	
+        if (isNativeAsync){
+            // process the response as a native async response
+            requestEndpoint.invokeAsyncResponse(responseMessage);
+        } else {
+            // process the response as a non-native async response
+            responseEndpointReference.invoke(responseMessage);
+        }
+    } // end method invokeAsyncReponse(Message)
+    
+    public T getResponseTargetAddress() {
+		return responseTargetAddress;
+	}
+
+	public void setResponseTargetAddress(T responseTargetAddress) {
+		this.responseTargetAddress = responseTargetAddress;
+	}
+
+	public String getRelatesToMsgID() {
+		return relatesToMsgID;
+	}
+
+	public void setRelatesToMsgID(String relatesToMsgID) {
+		this.relatesToMsgID = relatesToMsgID;
+	}
+
+	/**
+     * Invokes the async response where the parameter is Java bean(s) 
+     * - this method creates a Tuscany message
+     * 
+     * @param args the response data
+     * @param headers - any header 
+     */
+    public void invokeAsyncResponse(Object args, Map<String, Object> headers) {
+        
+        Message msg = messageFactory.createMessage();
+
+        msg.setOperation(getOperation( args ));
+        
+        // If this is not native async, then any Throwable is being passed as a parameter and
+        // requires wrapping
+        if( !isNativeAsync && args instanceof Throwable ) {
+        	args = new AsyncFaultWrapper( (Throwable) args ); 
+        } // end if
+        
+        // If this is not native async, then the message must contain an array of args since
+        // this is what is expected when invoking an EPR for the async response...
+        if( !isNativeAsync ) {
+        	Object[] objs = new Object[1];
+        	objs[0] = args;
+        	args = objs;
+        } // end if
+
+        msg.setTo(requestEndpoint);
+        msg.setFrom(responseEndpointReference);
+        
+        if( headers != null ) {
+        	msg.getHeaders().putAll(headers);
+        }
+        
+        if( args instanceof Throwable ) {
+        	msg.setFaultBody(args);
+        } else {
+        	msg.setBody(args);
+        } // end if
+        
+        invokeAsyncResponse(msg);
+        
+    } // end method invokeAsyncResponse(Object)
+
+	private Operation getOperation( Object args ) {
+		if( isNativeAsync ) {
+			List<Operation> ops = requestEndpoint.getService().getInterfaceContract().getInterface().getOperations();
+			for (Operation op : ops) {
+				if( operationName.equals(op.getName()) ) return op;
+			} // end for
+			return null;
+		} else {
+			operationName = "setResponse";
+			if( args instanceof Throwable ) { operationName = "setWrappedFault"; }
+			List<Operation> ops = responseEndpointReference.getReference().getInterfaceContract().getInterface().getOperations();
+			for (Operation op : ops) {
+				if( operationName.equals(op.getName()) ) return op;
+			} // end for
+			return null;
+		} // end if 
+	} // end getOperation
+
+	public void setBindingType(String bindingType) {
+		this.bindingType = bindingType;
+	} // end method setBindingType
+
+	public String getBindingType() {
+		return bindingType;
+	} // end method getBindingType
+
+    public RuntimeEndpoint getRequestEndpoint() {
+	return this.requestEndpoint;
+    }
+
+    public RuntimeEndpointReference getResponseEndpointReference() {
+	return this.responseEndpointReference;
+    }
+
+	public void setResponseEndpointReference(
+			RuntimeEndpointReference responseEndpointReference) {
+		this.responseEndpointReference = responseEndpointReference;
+    }
+	
+    @SuppressWarnings("unchecked")
+	private void readObject(ObjectInputStream in) throws IOException, ClassNotFoundException
+    {
+	    in.defaultReadObject();
+	    
+	    requestEndpoint = retrieveEndpoint(endpointURI);
+	    responseEndpointReference = retrieveEndpointReference(endpointReferenceURI);
+	    
+	    messageFactory = getMessageFactory();
+	    
+	    if (responseTargetAddress instanceof EndpointReference){
+	        // fix the target as in this case it will be an EPR
+	        EndpointReference epr = (EndpointReference)responseTargetAddress;
+	        responseTargetAddress = (T)retrieveEndpointReference(epr.getURI());
+	    } // end if    
+    } // end method readObject	
+
+    /**
+     * Gets a message factory
+     * @return
+     */
+    private MessageFactory getMessageFactory() {
+    	return registry.getExtensionPoint(FactoryExtensionPoint.class).getFactory(MessageFactory.class);
+	} // end method getMessageFactory
+
+	/**
+     * Fetches the EndpointReference identified by an endpoint reference URI
+     * @param uri - the URI of the endpoint reference
+     * @return - the EndpointReference matching the supplied URI - null if no EPR is found which
+     * matches the URI
+     */
+    private RuntimeEndpointReference retrieveEndpointReference(String uri) {
+		if( uri == null ) return null;
+    	if( domainRegistry == null ) return null;
+		List<EndpointReference> refs = domainRegistry.findEndpointReferences( uri );
+		// If there is more than EndpointReference with the uri...
+		if( refs.isEmpty() ) return null;
+		// TODO: what if there is more than 1 EPR with the given URI?
+		return (RuntimeEndpointReference) refs.get(0);
+	} // end method retrieveEndpointReference
+
+	/**
+     * Fetches the Endpoint identified by an endpoint URI
+     * - the Endpoint is retrieved from the DomainRegistry 
+     * @param uri - the URI of the Endpoint
+     * @return - the Endpoint corresponding to the URI, or null if no Endpoint is found which has the
+     * supplied URI
+     */
+	private RuntimeEndpoint retrieveEndpoint(String uri) {
+		if( uri == null ) return null;
+		if( domainRegistry == null ) domainRegistry = getEndpointRegistry( uri );
+		if( domainRegistry == null ) return null;
+		// TODO what if more than one Endpoint gets returned??
+		return (RuntimeEndpoint) domainRegistry.findEndpoint(uri).get(0);
+	} // end method retrieveEndpoint
+
+	/**
+	 * Gets the DomainRegistry which contains an Endpoint with the supplied URI
+	 * @param uri - The URI of an Endpoint
+	 * @return - the DomainRegistry containing the Endpoint with the supplied URI - null if no
+	 *           such DomainRegistry can be found
+	 */
+	private DomainRegistry getEndpointRegistry(String uri) {
+		ExtensionPointRegistry registry   = null;
+		DomainRegistry domainRegistry = null;
+		
+		CompositeContext context = CompositeContext.getCurrentCompositeContext();
+		if( context == null && requestEndpoint != null ) context = requestEndpoint.getCompositeContext();
+		if( context != null ) {
+			registry = context.getExtensionPointRegistry();
+			domainRegistry = getEndpointRegistry( registry );
+			if( domainRegistry != null ) {
+				this.registry = registry;
+				return domainRegistry;
+			} // end if
+		} // end if
+		
+		// Deal with the case where there is no context available
+    	for(ExtensionPointRegistry r : ExtensionPointRegistryLocator.getExtensionPointRegistries()) {
+                registry = r;
+    		if( registry != null ) {
+    			// Find the actual Endpoint in the DomainRegistry
+        		domainRegistry = getEndpointRegistry( registry );
+                
+                if( domainRegistry != null ) {
+                    for( Endpoint endpoint : domainRegistry.findEndpoint(uri) ) {
+                    	// TODO: For the present, simply return the first registry with a matching endpoint
+                    	this.registry = registry;
+                    	return domainRegistry;
+                    } // end for
+                } // end if 
+    		} // end if
+        } // end for
+		
+		return null;
+	} // end method getEndpointRegistry
+	
+    /**
+     * Get the DomainRegistry
+     * @param registry - the ExtensionPoint registry
+     * @return the DomainRegistry - will be null if the DomainRegistry cannot be found
+     */
+    private DomainRegistry getEndpointRegistry( ExtensionPointRegistry registry) {
+        DomainRegistryFactory domainRegistryFactory = ExtensibleDomainRegistryFactory.getInstance(registry);
+        
+        if( domainRegistryFactory == null ) return null;
+        
+        // Find the first endpoint registry that matches the domain name 
+        if( domainURI != null ) {
+	        for( DomainRegistry domainRegistry : domainRegistryFactory.getEndpointRegistries() ) {
+	        	if( domainURI.equals( domainRegistry.getDomainURI() ) ) return domainRegistry;
+	        } // end for
+        } // end if
+        
+        // if there was no domainName to match, simply return the first DomainRegistry if there is one...
+        
+        if (domainRegistryFactory.getEndpointRegistries().size() > 0){
+            DomainRegistry domainRegistry = (DomainRegistry) domainRegistryFactory.getEndpointRegistries().toArray()[0];
+            return domainRegistry;
+        } else {
+            return null;
+        }
+        
+    } // end method 
+	
+} // end class