Author: edwardsmj
Date: Mon Jan 24 14:57:26 2011
New Revision: 1062814
URL: http://svn.apache.org/viewvc?rev=1062814&view=rev
Log:
Fixing serialization of AsyncResponseInvoker in support of async services under
TUSCANY-3783
Modified:
tuscany/sca-java-2.x/trunk/modules/core-spi/src/main/java/org/apache/tuscany/sca/runtime/BaseEndpointRegistry.java
tuscany/sca-java-2.x/trunk/modules/core-spi/src/main/java/org/apache/tuscany/sca/runtime/EndpointRegistry.java
tuscany/sca-java-2.x/trunk/modules/core/src/main/java/org/apache/tuscany/sca/core/invocation/AsyncResponseInvoker.java
tuscany/sca-java-2.x/trunk/modules/implementation-java-runtime/src/main/java/org/apache/tuscany/sca/implementation/java/invocation/ResponseDispatchImpl.java
Modified:
tuscany/sca-java-2.x/trunk/modules/core-spi/src/main/java/org/apache/tuscany/sca/runtime/BaseEndpointRegistry.java
URL:
http://svn.apache.org/viewvc/tuscany/sca-java-2.x/trunk/modules/core-spi/src/main/java/org/apache/tuscany/sca/runtime/BaseEndpointRegistry.java?rev=1062814&r1=1062813&r2=1062814&view=diff
==============================================================================
---
tuscany/sca-java-2.x/trunk/modules/core-spi/src/main/java/org/apache/tuscany/sca/runtime/BaseEndpointRegistry.java
(original)
+++
tuscany/sca-java-2.x/trunk/modules/core-spi/src/main/java/org/apache/tuscany/sca/runtime/BaseEndpointRegistry.java
Mon Jan 24 14:57:26 2011
@@ -111,6 +111,23 @@ public abstract class BaseEndpointRegist
public List<EndpointReference> findEndpointReference(Endpoint endpoint) {
return endpointreferences;
}
+
+ /**
+ * Returns a list of EndpointReferences that have a URI that matches a
given URI
+ * @param uri - the URI to match
+ * @return a List of EndpointReferences that match the supplied URI - if
there are none
+ * an *empty* list is returned (not null)
+ */
+ public List<EndpointReference> findEndpointReferences( String uri ) {
+ List<EndpointReference> theRefs = new ArrayList<EndpointReference>();
+ if( uri == null ) return theRefs;
+
+ for( EndpointReference ref : endpointreferences ) {
+ if( uri.equals(ref.getURI()) ) theRefs.add(ref);
+ } // end for
+
+ return theRefs;
+ } // end method findEndpointReference
public abstract Endpoint getEndpoint(String uri);
Modified:
tuscany/sca-java-2.x/trunk/modules/core-spi/src/main/java/org/apache/tuscany/sca/runtime/EndpointRegistry.java
URL:
http://svn.apache.org/viewvc/tuscany/sca-java-2.x/trunk/modules/core-spi/src/main/java/org/apache/tuscany/sca/runtime/EndpointRegistry.java?rev=1062814&r1=1062813&r2=1062814&view=diff
==============================================================================
---
tuscany/sca-java-2.x/trunk/modules/core-spi/src/main/java/org/apache/tuscany/sca/runtime/EndpointRegistry.java
(original)
+++
tuscany/sca-java-2.x/trunk/modules/core-spi/src/main/java/org/apache/tuscany/sca/runtime/EndpointRegistry.java
Mon Jan 24 14:57:26 2011
@@ -63,6 +63,14 @@ public interface EndpointRegistry {
void removeEndpointReference(EndpointReference endpointReference);
// List<EndpointReference> findEndpointReference(Endpoint endpoint);
List<EndpointReference> getEndpointReferences();
+
+ /**
+ * Returns a list of EndpointReferences that have a URI that matches a
given URI
+ * @param uri - the URI to match
+ * @return a List of EndpointReferences that match the supplied URI - if
there are none
+ * an *empty* list is returned (not null)
+ */
+ public List<EndpointReference> findEndpointReferences( String uri );
void addListener(EndpointListener listener);
void removeListener(EndpointListener listener);
Modified:
tuscany/sca-java-2.x/trunk/modules/core/src/main/java/org/apache/tuscany/sca/core/invocation/AsyncResponseInvoker.java
URL:
http://svn.apache.org/viewvc/tuscany/sca-java-2.x/trunk/modules/core/src/main/java/org/apache/tuscany/sca/core/invocation/AsyncResponseInvoker.java?rev=1062814&r1=1062813&r2=1062814&view=diff
==============================================================================
---
tuscany/sca-java-2.x/trunk/modules/core/src/main/java/org/apache/tuscany/sca/core/invocation/AsyncResponseInvoker.java
(original)
+++
tuscany/sca-java-2.x/trunk/modules/core/src/main/java/org/apache/tuscany/sca/core/invocation/AsyncResponseInvoker.java
Mon Jan 24 14:57:26 2011
@@ -25,12 +25,16 @@ import java.io.Serializable;
import java.util.Collection;
import java.util.Iterator;
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.FactoryExtensionPoint;
+import org.apache.tuscany.sca.core.UtilityExtensionPoint;
+import org.apache.tuscany.sca.core.assembly.impl.RuntimeEndpointImpl;
+import org.apache.tuscany.sca.interfacedef.InterfaceContractMapper;
import org.apache.tuscany.sca.interfacedef.Operation;
import org.apache.tuscany.sca.invocation.Invoker;
import org.apache.tuscany.sca.invocation.InvokerAsyncResponse;
@@ -40,6 +44,7 @@ import org.apache.tuscany.sca.node.NodeF
import org.apache.tuscany.sca.provider.EndpointAsyncProvider;
import org.apache.tuscany.sca.runtime.DomainRegistryFactory;
import org.apache.tuscany.sca.runtime.EndpointRegistry;
+import org.apache.tuscany.sca.runtime.EndpointSerializer;
import org.apache.tuscany.sca.runtime.ExtensibleDomainRegistryFactory;
import org.apache.tuscany.sca.runtime.RuntimeEndpoint;
import org.apache.tuscany.sca.runtime.RuntimeEndpointReference;
@@ -60,8 +65,8 @@ public class AsyncResponseInvoker<T> imp
*/
private static final long serialVersionUID = -7992598227671386588L;
- private RuntimeEndpoint requestEndpoint;
- private RuntimeEndpointReference responseEndpointReference;
+ private transient RuntimeEndpoint requestEndpoint;
+ private transient RuntimeEndpointReference responseEndpointReference;
private T responseTargetAddress;
private String relatesToMsgID;
private String operationName;
@@ -69,6 +74,12 @@ public class AsyncResponseInvoker<T> imp
private String bindingType = "";
private boolean isNativeAsync;
+ private String endpointURI;
+ private String endpointReferenceURI;
+ private String domainURI;
+
+ private transient EndpointRegistry endpointRegistry;
+
public AsyncResponseInvoker(RuntimeEndpoint requestEndpoint,
RuntimeEndpointReference responseEndpointReference,
T responseTargetAddress, String relatesToMsgID,
@@ -81,6 +92,20 @@ public class AsyncResponseInvoker<T> imp
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();
+ } // end if
+
if ((requestEndpoint.getBindingProvider() instanceof
EndpointAsyncProvider) &&
(((EndpointAsyncProvider)requestEndpoint.getBindingProvider()).supportsNativeAsync())){
isNativeAsync = true;
@@ -126,8 +151,9 @@ public class AsyncResponseInvoker<T> imp
* - this method creates a Tuscany message
*
* @param args the response data
+ * @param headers - any header
*/
- public void invokeAsyncResponse(Object args) {
+ public void invokeAsyncResponse(Object args, Map<String, Object> headers) {
Message msg = messageFactory.createMessage();
@@ -150,6 +176,10 @@ public class AsyncResponseInvoker<T> imp
msg.setTo(requestEndpoint);
msg.setFrom(responseEndpointReference);
+ if( headers != null ) {
+ msg.getHeaders().putAll(headers);
+ }
+
if( args instanceof Throwable ) {
msg.setFaultBody(args);
} else {
@@ -199,27 +229,111 @@ public class AsyncResponseInvoker<T> imp
this.responseEndpointReference = responseEndpointReference;
}
- private void readObject(ObjectInputStream in) throws IOException,
ClassNotFoundException
+ @SuppressWarnings("unchecked")
+ private void readObject(ObjectInputStream in) throws IOException,
ClassNotFoundException
{
in.defaultReadObject();
- // find the real endpoint
- ExtensionPointRegistry extensionPointRegistry =
NodeFactory.getInstance().getExtensionPointRegistry();
- DomainRegistryFactory domainRegistryFactory =
ExtensibleDomainRegistryFactory.getInstance(extensionPointRegistry);
- Collection<EndpointRegistry> endpointRegistries =
domainRegistryFactory.getEndpointRegistries();
- EndpointRegistry endpointRegistry =
endpointRegistries.iterator().next();
- List<Endpoint> endpoints =
endpointRegistry.findEndpoint(requestEndpoint.getURI());
- requestEndpoint = (RuntimeEndpoint)endpoints.get(0);
+ requestEndpoint = retrieveEndpoint(endpointURI);
+ responseEndpointReference =
retrieveEndpointReference(endpointReferenceURI);
if (responseTargetAddress instanceof EndpointReference){
// fix the target as in this case it will be an EPR
EndpointReference epr =
(EndpointReference)responseTargetAddress;
- List<EndpointReference> endpointReferences =
endpointRegistry.getEndpointReferences();
- for (EndpointReference endpointReference : endpointReferences){
- if (endpointReference.getURI().equals(epr.getURI())){
- responseTargetAddress = (T)endpointReference;
- }
- }
- }
- }
+ responseTargetAddress =
(T)retrieveEndpointReference(epr.getURI());
+ } // end if
+ } // end method readObject
+
+ /**
+ * 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( endpointRegistry == null ) return null;
+ List<EndpointReference> refs =
endpointRegistry.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 EndpointRegistry
+ * @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( endpointRegistry == null ) endpointRegistry =
getEndpointRegistry( uri );
+ if( endpointRegistry == null ) return null;
+ // TODO what if more than one Endpoint gets returned??
+ return (RuntimeEndpoint)
endpointRegistry.findEndpoint(uri).get(0);
+ } // end method retrieveEndpoint
+
+ /**
+ * Gets the EndpointRegistry which contains an Endpoint with the
supplied URI
+ * @param uri - The URI of an Endpoint
+ * @return - the EndpointRegistry containing the Endpoint with the
supplied URI - null if no
+ * such EndpointRegistry can be found
+ */
+ private EndpointRegistry getEndpointRegistry(String uri) {
+ ExtensionPointRegistry registry = null;
+ EndpointRegistry endpointRegistry = null;
+
+ CompositeContext context =
CompositeContext.getCurrentCompositeContext();
+ if( context == null && requestEndpoint != null ) context =
requestEndpoint.getCompositeContext();
+ if( context != null ) {
+ registry = context.getExtensionPointRegistry();
+ endpointRegistry = getEndpointRegistry( registry );
+ if( endpointRegistry != null ) return endpointRegistry;
+ } // end if
+
+ // Deal with the case where there is no context available
+ for( NodeFactory factory : NodeFactory.getNodeFactories() ) {
+ registry = factory.getExtensionPointRegistry();
+ if( registry != null ) {
+ // Find the actual Endpoint in the EndpointRegistry
+ endpointRegistry = getEndpointRegistry( registry );
+
+ if( endpointRegistry != null ) {
+ for( Endpoint endpoint :
endpointRegistry.findEndpoint(uri) ) {
+ // TODO: For the present, simply return the first
registry with a matching endpoint
+ return endpointRegistry;
+ } // end for
+ } // end if
+ } // end if
+ } // end for
+
+ return null;
+ } // end method getEndpointRegistry
+
+ /**
+ * Get the EndpointRegistry
+ * @param registry - the ExtensionPoint registry
+ * @return the EndpointRegistry - will be null if the EndpointRegistry
cannot be found
+ */
+ private EndpointRegistry 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( EndpointRegistry endpointRegistry :
domainRegistryFactory.getEndpointRegistries() ) {
+ if( domainURI.equals( endpointRegistry.getDomainURI() )
) return endpointRegistry;
+ } // end for
+ } // end if
+
+ // if there was no domainName to match, simply return the first
EndpointRegistry...
+ EndpointRegistry endpointRegistry = (EndpointRegistry)
domainRegistryFactory.getEndpointRegistries().toArray()[0];
+
+
+ return endpointRegistry;
+ } // end method
+
} // end class
Modified:
tuscany/sca-java-2.x/trunk/modules/implementation-java-runtime/src/main/java/org/apache/tuscany/sca/implementation/java/invocation/ResponseDispatchImpl.java
URL:
http://svn.apache.org/viewvc/tuscany/sca-java-2.x/trunk/modules/implementation-java-runtime/src/main/java/org/apache/tuscany/sca/implementation/java/invocation/ResponseDispatchImpl.java?rev=1062814&r1=1062813&r2=1062814&view=diff
==============================================================================
---
tuscany/sca-java-2.x/trunk/modules/implementation-java-runtime/src/main/java/org/apache/tuscany/sca/implementation/java/invocation/ResponseDispatchImpl.java
(original)
+++
tuscany/sca-java-2.x/trunk/modules/implementation-java-runtime/src/main/java/org/apache/tuscany/sca/implementation/java/invocation/ResponseDispatchImpl.java
Mon Jan 24 14:57:26 2011
@@ -138,7 +138,7 @@ public class ResponseDispatchImpl<T> imp
// Use response invoker if present
if( respInvoker != null ) {
//respInvoker.invokeAsyncResponse(new
AsyncFaultWrapper(e));
- respInvoker.invokeAsyncResponse(e);
+ respInvoker.invokeAsyncResponse(e, null);
return;
} // end if
@@ -153,6 +153,7 @@ public class ResponseDispatchImpl<T> imp
* @throws IllegalStateException if either the sendResponse method or
the sendFault method have been called previously
* @param res - the response message, which is of type T
*/
+ @SuppressWarnings("unchecked")
public void sendResponse(T res) {
if( sendOK() ) {
lock.lock();
@@ -175,7 +176,7 @@ public class ResponseDispatchImpl<T> imp
// Use response invoker if present
if( respInvoker != null ) {
- respInvoker.invokeAsyncResponse(res);
+ respInvoker.invokeAsyncResponse(res, null);
return;
} // end if
} // end method sendResponse
