Stage 1 of Axis2 binding-ws support of async callbacks, as described in TUSCANY-3821

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

2 files changed, 83 insertions, 12 deletions

diff --git a/sca-java-2.x/trunk/modules/binding-ws-runtime-axis2/src/main/java/org/apache/tuscany/sca/binding/ws/axis2/provider/Axis2ReferenceBindingInvoker.java b/sca-java-2.x/trunk/modules/binding-ws-runtime-axis2/src/main/java/org/apache/tuscany/sca/binding/ws/axis2/provider/Axis2ReferenceBindingInvoker.java
index 32f6ee9ca5..3f3e63951b 100644
--- a/sca-java-2.x/trunk/modules/binding-ws-runtime-axis2/src/main/java/org/apache/tuscany/sca/binding/ws/axis2/provider/Axis2ReferenceBindingInvoker.java
+++ b/sca-java-2.x/trunk/modules/binding-ws-runtime-axis2/src/main/java/org/apache/tuscany/sca/binding/ws/axis2/provider/Axis2ReferenceBindingInvoker.java
@@ -49,13 +49,23 @@ import org.apache.axis2.client.OperationClient;
 import org.apache.axis2.client.Options;
 import org.apache.axis2.client.ServiceClient;
 import org.apache.axis2.context.MessageContext;
+import org.apache.tuscany.sca.assembly.AssemblyFactory;
 import org.apache.tuscany.sca.assembly.ComponentReference;
 import org.apache.tuscany.sca.assembly.Endpoint;
 import org.apache.tuscany.sca.binding.ws.WebServiceBinding;
+import org.apache.tuscany.sca.binding.ws.WebServiceBindingFactory;
+import org.apache.tuscany.sca.context.CompositeContext;
+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.invocation.AsyncResponseInvoker;
+import org.apache.tuscany.sca.core.invocation.Constants;
 import org.apache.tuscany.sca.interfacedef.util.FaultException;
 import org.apache.tuscany.sca.interfacedef.wsdl.WSDLInterface;
 import org.apache.tuscany.sca.invocation.Invoker;
 import org.apache.tuscany.sca.invocation.Message;
+import org.apache.tuscany.sca.invocation.MessageFactory;
+import org.apache.tuscany.sca.runtime.RuntimeEndpoint;
 import org.apache.tuscany.sca.runtime.RuntimeEndpointReference;
 import org.oasisopen.sca.ServiceRuntimeException;
 
@@ -209,9 +219,16 @@ public class Axis2ReferenceBindingInvoker implements Invoker {
         // Axis2 operationClients can not be shared so create a new one for each request
         final OperationClient operationClient = serviceClient.createClient(wsdlOperationName);
         operationClient.setOptions(options);
-
-        Endpoint callbackEndpoint = msg.getFrom().getCallbackEndpoint();
-
+        
+        Endpoint callbackEndpoint;
+        AsyncResponseInvoker<String> respInvoker = (AsyncResponseInvoker<String>) msg.getHeaders().get(Constants.ASYNC_RESPONSE_INVOKER);
+        if( respInvoker != null ) {
+        	callbackEndpoint = createAsyncResponseEndpoint( msg, respInvoker );
+        	msg.setTo(callbackEndpoint);
+        } else {
+        	callbackEndpoint = msg.getFrom().getCallbackEndpoint();
+        } // end if 
+        
         SOAPEnvelope sev = requestMC.getEnvelope();
         SOAPHeader sh = sev.getHeader();
         
@@ -251,7 +268,36 @@ public class Axis2ReferenceBindingInvoker implements Invoker {
         return operationClient;
     } // end method createOperationClient
     
-    private String getToAddress( Message msg ) throws ServiceRuntimeException {
+    /**
+     * Create an Async Response Endpoint
+     * @param msg - the Tuscany message
+     * @param respInvoker - the AsyncResponseInvoker for the async response
+     * @return - an Endpoint which embodies the callback address
+     */
+    private Endpoint createAsyncResponseEndpoint(Message msg,
+			AsyncResponseInvoker<String> respInvoker) {
+		String callbackAddress = respInvoker.getResponseTargetAddress();
+		if( callbackAddress == null ) return null;
+		
+		// Get the necessary factories
+		ExtensionPointRegistry registry = endpointReference.getCompositeContext().getExtensionPointRegistry();
+        FactoryExtensionPoint modelFactories = registry.getExtensionPoint(FactoryExtensionPoint.class);
+        RuntimeAssemblyFactory assemblyFactory = (RuntimeAssemblyFactory)modelFactories.getFactory(AssemblyFactory.class);
+        WebServiceBindingFactory webServiceBindingFactory = (WebServiceBindingFactory)modelFactories.getFactory(WebServiceBindingFactory.class);
+		
+        // Create the endpoint
+        RuntimeEndpoint callbackEndpoint = (RuntimeEndpoint)assemblyFactory.createEndpoint();
+        // Add a binding
+        WebServiceBinding cbBinding = webServiceBindingFactory.createWebServiceBinding();
+        cbBinding.setURI(callbackAddress);
+        callbackEndpoint.setBinding(cbBinding);
+        // Embed the response Address URI
+        callbackEndpoint.setURI(callbackAddress);
+        callbackEndpoint.setUnresolved(true);
+		return callbackEndpoint;
+	} // end method createAsyncResponseEndpoint
+
+	private String getToAddress( Message msg ) throws ServiceRuntimeException {
     	String address = null;
     	
         // if target endpoint was not specified when this invoker was created, 
@@ -292,11 +338,11 @@ public class Axis2ReferenceBindingInvoker implements Invoker {
      * @throws AxisFault - if an error occurs setting the wsa:From into the header
      */
     private void addWSAMessageIDHeader( SOAPHeader sh, String msgID ) throws AxisFault {
+    	if( msgID == null ) return;
         OMElement idHeader = sh.getOMFactory().createOMElement(QNAME_WSA_MESSAGEID);
         idHeader.setText( msgID );
         
         sh.addChild(idHeader);
-
     } // end method addWSAMessageIDHeader
     
     private static String WS_REF_PARMS = "WS_REFERENCE_PARAMETERS";
@@ -353,7 +399,7 @@ public class Axis2ReferenceBindingInvoker implements Invoker {
      * @param msg - the message
      */
     private void addWSARelatesTo( SOAPHeader sh, Message msg ) {
-    	String idValue = (String) msg.getHeaders().get(WS_MESSAGE_ID);
+    	String idValue = (String) msg.getHeaders().get("RELATES_TO");
     	if( idValue != null ){
             OMElement relatesToOM = sh.getOMFactory().createOMElement( QNAME_WSA_RELATESTO );
             OMAttribute relType = sh.getOMFactory().createOMAttribute("RelationshipType", null, SCA_CALLBACK_REL);
diff --git a/sca-java-2.x/trunk/modules/binding-ws-runtime-axis2/src/main/java/org/apache/tuscany/sca/binding/ws/axis2/provider/TuscanyServiceProvider.java b/sca-java-2.x/trunk/modules/binding-ws-runtime-axis2/src/main/java/org/apache/tuscany/sca/binding/ws/axis2/provider/TuscanyServiceProvider.java
index ba23e80d88..70875916df 100644
--- a/sca-java-2.x/trunk/modules/binding-ws-runtime-axis2/src/main/java/org/apache/tuscany/sca/binding/ws/axis2/provider/TuscanyServiceProvider.java
+++ b/sca-java-2.x/trunk/modules/binding-ws-runtime-axis2/src/main/java/org/apache/tuscany/sca/binding/ws/axis2/provider/TuscanyServiceProvider.java
@@ -46,10 +46,14 @@ import org.apache.tuscany.sca.binding.ws.WebServiceBindingFactory;
 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.RuntimeEndpointImpl;

+import org.apache.tuscany.sca.core.invocation.AsyncResponseInvoker;

+import org.apache.tuscany.sca.core.invocation.Constants;

 import org.apache.tuscany.sca.interfacedef.Operation;

 import org.apache.tuscany.sca.invocation.Message;

 import org.apache.tuscany.sca.invocation.MessageFactory;

 import org.apache.tuscany.sca.runtime.RuntimeEndpoint;

+import org.apache.tuscany.sca.runtime.RuntimeEndpointReference;

 import org.oasisopen.sca.ServiceRuntimeException;

 

 public class TuscanyServiceProvider {

@@ -151,6 +155,9 @@ public class TuscanyServiceProvider {
 

         // Create a from EPR to hold the details of the callback endpoint, if any

         createCallbackEPR( callbackAddress, inMC, msg );

+        

+        // Set up any async response EPR

+        setupAsyncResponse( msg, callbackAddress );

 

         Message response = endpoint.invoke(msg);

         

@@ -199,6 +206,26 @@ public class TuscanyServiceProvider {
     } // end method 

     

     /**

+     * Setup the necessary infrastructure for the Async response handling

+     * @param msg

+     * @param callbackAddress 

+     */

+    private void setupAsyncResponse(Message msg, String callbackAddress) {

+    	if( !endpoint.isAsyncInvocation() ) return;

+

+    	endpoint.createAsyncServerCallback();

+    	RuntimeEndpointReference asyncCallback = endpoint.getAsyncServerCallback();

+    	

+    	// Create a response invoker, containing the callback address and add it to the message headers

+        AsyncResponseInvoker<String> respInvoker = 

+        	new AsyncResponseInvoker<String>(endpoint, asyncCallback,

+        			callbackAddress, 

+        			(String)msg.getHeaders().get(Constants.MESSAGE_ID),

+        			msg.getOperation().getName(), messageFactory);

+        msg.getHeaders().put(Constants.ASYNC_RESPONSE_INVOKER, respInvoker);

+	} // end method setupAsyncResponse

+

+	/**

      * If there is a callback address, create an EPR for the callback with a referenced endpoint that contains

      * the binding and the target callback address

      * @param callbackAddress - the callback address - may be null

@@ -216,7 +243,7 @@ public class TuscanyServiceProvider {
             from.setTargetEndpoint(fromEndpoint);

             from.setStatus(EndpointReference.Status.WIRED_TARGET_FOUND_AND_MATCHED);

             msg.setFrom(from);

-            Endpoint callbackEndpoint = assemblyFactory.createEndpoint();

+            RuntimeEndpoint callbackEndpoint = (RuntimeEndpoint)assemblyFactory.createEndpoint();

             //

             WebServiceBinding cbBinding = webServiceBindingFactory.createWebServiceBinding();

             cbBinding.setURI(callbackAddress);

@@ -262,7 +289,6 @@ public class TuscanyServiceProvider {
     	return callbackAddress;

     } // end method handleCallbackAddress

     

-    private static String WS_MESSAGE_ID = "WS_MESSAGE_ID";

     /**

      * Handle a SOAP wsa:MessageID header - place the contents into the Tuscany message for use by any callback

      * @param header - the SOAP Headers

@@ -274,11 +300,10 @@ public class TuscanyServiceProvider {
         if (messageID != null) {

         	String idValue = messageID.getText();

         	// Store the value of the message ID element into the message under "WS_MESSAGE_ID"...

-        	msg.getHeaders().put(WS_MESSAGE_ID, idValue);

+        	msg.getHeaders().put(Constants.MESSAGE_ID, idValue);

         } // end if

     } // end method handleMessageID

     

-    private static String WS_RELATES_TO = "WS_RELATES_TO";

     /**

      * Handle a SOAP wsa:RelatesTo header - place the contents into the Tuscany message for use by any callback

      * @param header - the SOAP Headers

@@ -289,8 +314,8 @@ public class TuscanyServiceProvider {
         OMElement messageID = header.getFirstChildWithName(QNAME_WSA_RELATESTO);

         if (messageID != null) {

         	String idValue = messageID.getText();

-        	// Store the value of the message ID element into the message under "WS_MESSAGE_ID"...

-        	msg.getHeaders().put(WS_MESSAGE_ID, idValue);

+        	// Store the value of the message ID element into the message under "RELATES_TO"...

+        	msg.getHeaders().put(Constants.RELATES_TO, idValue);

         } // end if

     } // end method handleMessageID

 } // end class AsyncResponseHandler