From e5b7380c874745c989d1816b8f552504f038e1bc Mon Sep 17 00:00:00 2001 From: lresende Date: Thu, 26 Sep 2013 20:33:20 +0000 Subject: 2.0 branch for possible maintenance release git-svn-id: http://svn.us.apache.org/repos/asf/tuscany@1526672 13f79535-47bb-0310-9956-ffa450edef68 --- .../invocation/JavaAsyncImplementationInvoker.java | 232 +++++++++++++++++++++ 1 file changed, 232 insertions(+) create mode 100644 sca-java-2.x/branches/2.0/modules/implementation-java-runtime/src/main/java/org/apache/tuscany/sca/implementation/java/invocation/JavaAsyncImplementationInvoker.java (limited to 'sca-java-2.x/branches/2.0/modules/implementation-java-runtime/src/main/java/org/apache/tuscany/sca/implementation/java/invocation/JavaAsyncImplementationInvoker.java') diff --git a/sca-java-2.x/branches/2.0/modules/implementation-java-runtime/src/main/java/org/apache/tuscany/sca/implementation/java/invocation/JavaAsyncImplementationInvoker.java b/sca-java-2.x/branches/2.0/modules/implementation-java-runtime/src/main/java/org/apache/tuscany/sca/implementation/java/invocation/JavaAsyncImplementationInvoker.java new file mode 100644 index 0000000000..759724484e --- /dev/null +++ b/sca-java-2.x/branches/2.0/modules/implementation-java-runtime/src/main/java/org/apache/tuscany/sca/implementation/java/invocation/JavaAsyncImplementationInvoker.java @@ -0,0 +1,232 @@ +/* + * 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.implementation.java.invocation; + +import java.lang.reflect.InvocationTargetException; +import java.lang.reflect.Method; + +import org.apache.tuscany.sca.core.factory.InstanceWrapper; +import org.apache.tuscany.sca.core.factory.ObjectCreationException; +import org.apache.tuscany.sca.core.invocation.AsyncResponseException; +import org.apache.tuscany.sca.interfacedef.DataType; +import org.apache.tuscany.sca.interfacedef.Operation; +import org.apache.tuscany.sca.interfacedef.java.JavaOperation; +import org.apache.tuscany.sca.invocation.InterceptorAsync; +import org.apache.tuscany.sca.invocation.Invoker; +import org.apache.tuscany.sca.invocation.InvokerAsyncResponse; +import org.apache.tuscany.sca.invocation.Message; +import org.apache.tuscany.sca.runtime.RuntimeComponent; +import org.apache.tuscany.sca.runtime.RuntimeComponentService; +import org.oasisopen.sca.ResponseDispatch; +import org.oasisopen.sca.ServiceRuntimeException; + +/** + * Responsible for asynchronously dispatching an invocation to a Java component + * implementation instance + * + */ +public class JavaAsyncImplementationInvoker extends JavaImplementationInvoker implements InterceptorAsync { + + public JavaAsyncImplementationInvoker(Operation operation, Method method, RuntimeComponent component, RuntimeComponentService service) { + super( operation, method, component, service); + assert method != null : "Operation method cannot be null"; + assert ((JavaOperation) operation).isAsyncServer() : "Operation must be async"; + } // end constructor + + public Message invoke(Message msg) { + Operation op = this.operation; + + Object payload = msg.getBody(); + + // Save the current thread context classloader + ClassLoader tccl = Thread.currentThread().getContextClassLoader(); + + try { + // The following call might create a new conversation, as a result, the msg.getConversationID() might + // return a new value + InstanceWrapper wrapper = scopeContainer.getWrapper(null); + + Object instance = wrapper.getInstance(); + + // Set the TCCL to the classloader used to load the implementation class + Thread.currentThread().setContextClassLoader(instance.getClass().getClassLoader()); + + // For an async server method, there is an extra input parameter, which is a DispatchResponse instance + // which is typed by the type of the response + Class responseType = op.getOutputType().getPhysical(); + ResponseDispatch dispatch = ResponseDispatchImpl.newInstance(responseType, msg ); + + Object ret; + Object[] payload2; + if (payload != null && !payload.getClass().isArray()) { + payload2 = new Object[2]; + payload2[0] = payload; + } else { + payload2 = new Object[ ((Object[])payload).length + 1 ]; + for( int i = 0; i < ((Object[])payload).length; i++) { + payload2[i] = ((Object[])payload)[i]; + } // end for + } + payload2[ payload2.length - 1 ] = dispatch; + + ret = method.invoke(instance, (Object[])payload2); + + throw new InvocationTargetException( new AsyncResponseException("AsyncResponse") ); + + } catch (InvocationTargetException e) { + Throwable cause = e.getTargetException(); + boolean isChecked = false; + for (DataType d : operation.getFaultTypes()) { + if (d.getPhysical().isInstance(cause)) { + isChecked = true; + msg.setFaultBody(cause); + break; + } + } // end for + + if (!isChecked) { + if (cause instanceof RuntimeException) { + throw (RuntimeException)cause; + } + if (cause instanceof Error) { + throw (Error)cause; + } else { + throw new ServiceRuntimeException(cause.getMessage(), cause); + } + } // end if + + } catch (ObjectCreationException e) { + throw new ServiceRuntimeException(e.getMessage(), e); + } catch (Exception e) { + msg.setFaultBody(e); + } finally { + // set the tccl + Thread.currentThread().setContextClassLoader(tccl); + } + return msg; + } // end method invoke + + protected Invoker next; + protected InvokerAsyncResponse previous; + + public void setNext(Invoker next) { + this.next = next; + } + + public Invoker getNext() { + return next; + } + + public void invokeAsyncRequest(Message msg) throws Throwable { + processRequest(msg); + } // end method invokeAsyncRequest + + public void invokeAsyncResponse(Message msg) { + msg = processResponse(msg); + InvokerAsyncResponse thePrevious = (InvokerAsyncResponse)getPrevious(); + if (thePrevious != null ) thePrevious.invokeAsyncResponse(msg); + } // end method invokeAsyncResponse + + public void setPrevious(InvokerAsyncResponse previous) { + this.previous = previous; + } + + public InvokerAsyncResponse getPrevious() { + return previous; + } + + public Message processRequest(Message msg) { + Operation op = this.operation; + Object payload = msg.getBody(); + + // Replace TCCL with the class loader used to load the java class as per SCA Spec + ClassLoader tccl = Thread.currentThread().getContextClassLoader(); + + try { + InstanceWrapper wrapper = scopeContainer.getWrapper(null); + Object instance = wrapper.getInstance(); + + // Set the TCCL to the classloader used to load the implementation class + Thread.currentThread().setContextClassLoader(instance.getClass().getClassLoader()); + + // For an async server method, there is an extra input parameter, which is a DispatchResponse instance + // which is typed by the type of the response + Class responseType = op.getOutputType().getPhysical(); + ResponseDispatch dispatch = ResponseDispatchImpl.newInstance(responseType, msg ); + + Object[] payload2; + if (payload != null && !payload.getClass().isArray()) { + payload2 = new Object[2]; + payload2[0] = payload; + } else { + payload2 = new Object[ ((Object[])payload).length + 1 ]; + for( int i = 0; i < ((Object[])payload).length; i++) { + payload2[i] = ((Object[])payload)[i]; + } // end for + } + payload2[ payload2.length - 1 ] = dispatch; + + method.invoke(instance, (Object[])payload2); + + } catch (InvocationTargetException e) { + Throwable cause = e.getTargetException(); + boolean isChecked = false; + for (DataType d : operation.getFaultTypes()) { + if (d.getPhysical().isInstance(cause)) { + isChecked = true; + // Ignore these errors since they should be returned asynchronously + break; + } + } // end for + + if (!isChecked) { + if (cause instanceof RuntimeException) { + throw (RuntimeException)cause; + } // end if + if (cause instanceof Error) { + throw (Error)cause; + } else { + throw new ServiceRuntimeException(cause.getMessage(), cause); + } // end if + } // end if + + } catch (Exception e) { + throw new ServiceRuntimeException(e.getMessage(), e); + } finally { + // set the tccl + Thread.currentThread().setContextClassLoader(tccl); + } + return msg; + } // end method processRequest + + public Message postProcessRequest(Message msg) { + return msg; + } + + public Message postProcessRequest(Message msg, Throwable e) + throws Throwable { + throw e; + } + + public Message processResponse(Message msg) { + return msg; + } // end method processResponse + +} // end class JavaAsyncImplementationInvoker -- cgit v1.2.3