Author: dkulp
Date: Thu Apr 5 18:53:12 2012
New Revision: 1310009
URL: http://svn.apache.org/viewvc?rev=1310009&view=rev
Log:
Introduce a new @UseAsyncMethod to allow selecting the usage of the
async variants on the server side when using JAX-WS.
Added:
cxf/trunk/api/src/main/java/org/apache/cxf/annotations/UseAsyncMethod.java
(with props)
cxf/trunk/rt/frontend/jaxws/src/main/java/org/apache/cxf/jaxws/ServerAsyncResponse.java
(with props)
Modified:
cxf/trunk/api/src/main/java/org/apache/cxf/service/invoker/AbstractInvoker.java
cxf/trunk/rt/frontend/jaxws/src/main/java/org/apache/cxf/jaxws/AbstractJAXWSMethodInvoker.java
cxf/trunk/systests/jaxws/src/test/java/org/apache/cxf/systest/jaxws/ClientServerTest.java
cxf/trunk/systests/jaxws/src/test/java/org/apache/cxf/systest/jaxws/Server.java
Added:
cxf/trunk/api/src/main/java/org/apache/cxf/annotations/UseAsyncMethod.java
URL:
http://svn.apache.org/viewvc/cxf/trunk/api/src/main/java/org/apache/cxf/annotations/UseAsyncMethod.java?rev=1310009&view=auto
==============================================================================
--- cxf/trunk/api/src/main/java/org/apache/cxf/annotations/UseAsyncMethod.java
(added)
+++ cxf/trunk/api/src/main/java/org/apache/cxf/annotations/UseAsyncMethod.java
Thu Apr 5 18:53:12 2012
@@ -0,0 +1,71 @@
+/**
+ * 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.cxf.annotations;
+
+import java.lang.annotation.Documented;
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+/**
+ * Instructs the runtime to dispatch using the async method
+ * on service if continuations are available. This only applies
+ * to the JAX-WS frontend at this time.
+ *
+ * Instead of calling the "X methodName(Y, Z...) method, it will
+ * call the "Future<?> methodName(Y, Z, ... AsyncHandler<MethodResponse>)"
+ * method passing in an AsyncHandler that you will need to call when
+ * the response is ready. An example would be:
+ *
+ * <pre>
+ * public Future<?> greetMeAsync(final String requestType,
+ * final AsyncHandler<GreetMeResponse>
asyncHandler) {
+ * final ServerAsyncResponse<GreetMeResponse> r = new
ServerAsyncResponse<GreetMeResponse>();
+ * new Thread() {
+ * public void run() {
+ * //do some work on a backgound thread to generate the response...
+ * GreetMeResponse resp = new GreetMeResponse();
+ * resp.setResponseType("Hello " + requestType);
+ * r.set(resp);
+ * asyncHandler.handleResponse(r);
+ * }
+ * } .start();
+ * return r;
+ * }
+ * </pre>
+ *
+ * The use of the org.apache.cxf.jaxws.ServerAsyncResponse class for the
response
+ * as shown above can simplify things and is recommended.
+ */
+@Documented
+@Retention(RetentionPolicy.RUNTIME)
+@Target({ ElementType.METHOD })
+public @interface UseAsyncMethod {
+
+ /**
+ * By default, if continuations are not available,
+ * it will use the non-async method. If you ALWAYS
+ * want the async method called, set this to true. However,
+ * that can cause threads to block.
+ * @return
+ */
+ boolean always() default false;
+}
+
Propchange:
cxf/trunk/api/src/main/java/org/apache/cxf/annotations/UseAsyncMethod.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange:
cxf/trunk/api/src/main/java/org/apache/cxf/annotations/UseAsyncMethod.java
------------------------------------------------------------------------------
svn:keywords = Rev Date
Modified:
cxf/trunk/api/src/main/java/org/apache/cxf/service/invoker/AbstractInvoker.java
URL:
http://svn.apache.org/viewvc/cxf/trunk/api/src/main/java/org/apache/cxf/service/invoker/AbstractInvoker.java?rev=1310009&r1=1310008&r2=1310009&view=diff
==============================================================================
---
cxf/trunk/api/src/main/java/org/apache/cxf/service/invoker/AbstractInvoker.java
(original)
+++
cxf/trunk/api/src/main/java/org/apache/cxf/service/invoker/AbstractInvoker.java
Thu Apr 5 18:53:12 2012
@@ -59,11 +59,6 @@ public abstract class AbstractInvoker im
throw new Fault(new Message("EXCEPTION_INVOKING_OBJECT", LOG,
"No binding operation info",
"unknown method", "unknown"));
}
-
-
- //Method m =
(Method)bop.getOperationInfo().getProperty(Method.class.getName());
- m = matchMethod(m, serviceObject);
-
List<Object> params = null;
if (o instanceof List) {
params = CastUtils.cast((List<?>)o);
@@ -71,12 +66,25 @@ public abstract class AbstractInvoker im
params = new MessageContentsList(o);
}
+ m = adjustMethodAndParams(m, exchange, params);
+
+ //Method m =
(Method)bop.getOperationInfo().getProperty(Method.class.getName());
+ m = matchMethod(m, serviceObject);
+
+
return invoke(exchange, serviceObject, m, params);
} finally {
releaseServiceObject(exchange, serviceObject);
}
}
+ protected Method adjustMethodAndParams(Method m,
+ Exchange ex,
+ List<Object> params) {
+ //nothing to do
+ return m;
+ }
+
protected Object invoke(Exchange exchange, final Object serviceObject,
Method m, List<Object> params) {
Object res;
try {
Modified:
cxf/trunk/rt/frontend/jaxws/src/main/java/org/apache/cxf/jaxws/AbstractJAXWSMethodInvoker.java
URL:
http://svn.apache.org/viewvc/cxf/trunk/rt/frontend/jaxws/src/main/java/org/apache/cxf/jaxws/AbstractJAXWSMethodInvoker.java?rev=1310009&r1=1310008&r2=1310009&view=diff
==============================================================================
---
cxf/trunk/rt/frontend/jaxws/src/main/java/org/apache/cxf/jaxws/AbstractJAXWSMethodInvoker.java
(original)
+++
cxf/trunk/rt/frontend/jaxws/src/main/java/org/apache/cxf/jaxws/AbstractJAXWSMethodInvoker.java
Thu Apr 5 18:53:12 2012
@@ -28,13 +28,18 @@ import java.util.List;
import java.util.Map;
import javax.activation.DataHandler;
+import javax.xml.ws.AsyncHandler;
+import javax.xml.ws.Response;
import javax.xml.ws.handler.MessageContext;
import javax.xml.ws.handler.MessageContext.Scope;
import javax.xml.ws.soap.SOAPFaultException;
+import org.apache.cxf.annotations.UseAsyncMethod;
import org.apache.cxf.attachment.AttachmentImpl;
import org.apache.cxf.binding.soap.SoapFault;
import org.apache.cxf.binding.soap.SoapMessage;
+import org.apache.cxf.continuations.Continuation;
+import org.apache.cxf.continuations.ContinuationProvider;
import org.apache.cxf.endpoint.Endpoint;
import org.apache.cxf.headers.Header;
import org.apache.cxf.helpers.CastUtils;
@@ -43,12 +48,15 @@ import org.apache.cxf.jaxws.context.Wrap
import org.apache.cxf.message.Attachment;
import org.apache.cxf.message.Exchange;
import org.apache.cxf.message.Message;
+import org.apache.cxf.message.MessageContentsList;
import org.apache.cxf.message.MessageImpl;
import org.apache.cxf.service.invoker.Factory;
import org.apache.cxf.service.invoker.FactoryInvoker;
import org.apache.cxf.service.invoker.SingletonFactory;
+import org.apache.cxf.service.model.BindingOperationInfo;
public abstract class AbstractJAXWSMethodInvoker extends FactoryInvoker {
+ private static final String ASYNC_METHOD =
"org.apache.cxf.jaxws.async.method";
public AbstractJAXWSMethodInvoker(final Object bean) {
super(new SingletonFactory(bean));
@@ -67,6 +75,111 @@ public abstract class AbstractJAXWSMetho
}
return null;
}
+ protected Method adjustMethodAndParams(Method m, Exchange ex, List<Object>
params) {
+
+ UseAsyncMethod uam = m.getAnnotation(UseAsyncMethod.class);
+ if (uam != null) {
+ BindingOperationInfo bop = ex.getBindingOperationInfo();
+ Method ret = bop.getProperty(ASYNC_METHOD, Method.class);
+ if (ret == null) {
+ Class<?> ptypes[] = new Class<?>[m.getParameterTypes().length
+ 1];
+ System.arraycopy(m.getParameterTypes(), 0, ptypes, 0,
m.getParameterTypes().length);
+ ptypes[m.getParameterTypes().length] = AsyncHandler.class;
+ try {
+ ret = m.getDeclaringClass().getMethod(m.getName() +
"Async", ptypes);
+ bop.setProperty(ASYNC_METHOD, ret);
+ } catch (Throwable t) {
+ //ignore
+ }
+ }
+ if (ret != null) {
+ JaxwsServerHandler h = ex.get(JaxwsServerHandler.class);
+ if (h != null) {
+ return ret;
+ }
+ ContinuationProvider cp =
ex.getInMessage().get(ContinuationProvider.class);
+ if (cp == null && uam.always()) {
+ JaxwsServerHandler handler = new JaxwsServerHandler(null);
+ ex.put(JaxwsServerHandler.class, handler);
+ params.add(handler);
+ return ret;
+ } else if (cp != null && cp.getContinuation() != null) {
+ final Continuation c = cp.getContinuation();
+ c.suspend(0);
+ JaxwsServerHandler handler = new JaxwsServerHandler(c);
+ ex.put(JaxwsServerHandler.class, handler);
+ params.add(handler);
+ return ret;
+ }
+ }
+ }
+ return m;
+ }
+
+ class JaxwsServerHandler implements AsyncHandler<Object> {
+ Response<Object> r;
+ Continuation continuation;
+ boolean done;
+
+ public JaxwsServerHandler(Continuation c) {
+ continuation = c;
+ }
+
+ public synchronized void handleResponse(Response<Object> res) {
+ r = res;
+ done = true;
+ if (continuation != null) {
+ continuation.resume();
+ }
+ notifyAll();
+ }
+ public boolean hasContinuation() {
+ return continuation != null;
+ }
+ public synchronized void waitForDone() {
+ while (!done) {
+ try {
+ wait();
+ } catch (InterruptedException e) {
+ //ignore
+ }
+ }
+ }
+ public boolean isDone() {
+ return done;
+ }
+ public Object getObject() {
+ try {
+ return r.get();
+ } catch (Exception ex) {
+ throw new Fault(ex);
+ }
+ }
+ }
+
+ @Override
+ protected Object invoke(Exchange exchange,
+ final Object serviceObject, Method m,
+ List<Object> params) {
+ JaxwsServerHandler h = exchange.get(JaxwsServerHandler.class);
+ if (h != null && h.isDone()) {
+ BindingOperationInfo bop = exchange.getBindingOperationInfo();
+ if (bop.isUnwrapped()) {
+ exchange.put(BindingOperationInfo.class,
bop.getWrappedOperation());
+ }
+ return new MessageContentsList(h.getObject());
+ }
+ Object o = super.invoke(exchange, serviceObject, m, params);
+ if (h != null && !h.hasContinuation()) {
+ h.waitForDone();
+ BindingOperationInfo bop = exchange.getBindingOperationInfo();
+ if (bop.isUnwrapped()) {
+ exchange.put(BindingOperationInfo.class,
bop.getWrappedOperation());
+ }
+ return new MessageContentsList(h.getObject());
+ }
+ return o;
+ }
@Override
protected Fault createFault(Throwable ex, Method m, List<Object> params,
boolean checked) {
Added:
cxf/trunk/rt/frontend/jaxws/src/main/java/org/apache/cxf/jaxws/ServerAsyncResponse.java
URL:
http://svn.apache.org/viewvc/cxf/trunk/rt/frontend/jaxws/src/main/java/org/apache/cxf/jaxws/ServerAsyncResponse.java?rev=1310009&view=auto
==============================================================================
---
cxf/trunk/rt/frontend/jaxws/src/main/java/org/apache/cxf/jaxws/ServerAsyncResponse.java
(added)
+++
cxf/trunk/rt/frontend/jaxws/src/main/java/org/apache/cxf/jaxws/ServerAsyncResponse.java
Thu Apr 5 18:53:12 2012
@@ -0,0 +1,67 @@
+/**
+ * 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.cxf.jaxws;
+
+import java.util.Map;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+/**
+ *
+ */
+public class ServerAsyncResponse<T> implements javax.xml.ws.Response<T> {
+ T value;
+ boolean done;
+
+ /**
+ * Currently unused
+ */
+ public boolean cancel(boolean mayInterruptIfRunning) {
+ return false;
+ }
+ /**
+ * Currently unused
+ */
+ public boolean isCancelled() {
+ return false;
+ }
+ public boolean isDone() {
+ return done;
+ }
+ public void set(T t) {
+ value = t;
+ done = true;
+ }
+ public T get() throws InterruptedException, ExecutionException {
+ return value;
+ }
+ public T get(long timeout, TimeUnit unit)
+ throws InterruptedException, ExecutionException, TimeoutException {
+ return value;
+ }
+
+ /**
+ * Currently unused
+ */
+ public Map<String, Object> getContext() {
+ return null;
+ }
+}
Propchange:
cxf/trunk/rt/frontend/jaxws/src/main/java/org/apache/cxf/jaxws/ServerAsyncResponse.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange:
cxf/trunk/rt/frontend/jaxws/src/main/java/org/apache/cxf/jaxws/ServerAsyncResponse.java
------------------------------------------------------------------------------
svn:keywords = Rev Date
Modified:
cxf/trunk/systests/jaxws/src/test/java/org/apache/cxf/systest/jaxws/ClientServerTest.java
URL:
http://svn.apache.org/viewvc/cxf/trunk/systests/jaxws/src/test/java/org/apache/cxf/systest/jaxws/ClientServerTest.java?rev=1310009&r1=1310008&r2=1310009&view=diff
==============================================================================
---
cxf/trunk/systests/jaxws/src/test/java/org/apache/cxf/systest/jaxws/ClientServerTest.java
(original)
+++
cxf/trunk/systests/jaxws/src/test/java/org/apache/cxf/systest/jaxws/ClientServerTest.java
Thu Apr 5 18:53:12 2012
@@ -1025,5 +1025,15 @@ public class ClientServerTest extends Ab
InputStream ins = url.openStream();
ins.close();
}
-
+
+ @Test
+ public void testServerAsync() throws Exception {
+ Service service = Service.create(serviceName);
+ service.addPort(fakePortName,
javax.xml.ws.soap.SOAPBinding.SOAP11HTTP_BINDING,
+ "http://localhost:"; + PORT + "/SoapContext/AsyncSoapPort");
+ Greeter greeter = service.getPort(fakePortName, Greeter.class);
+ String resp = greeter.greetMe("World");
+ assertEquals("Hello World", resp);
+ }
+
}
Modified:
cxf/trunk/systests/jaxws/src/test/java/org/apache/cxf/systest/jaxws/Server.java
URL:
http://svn.apache.org/viewvc/cxf/trunk/systests/jaxws/src/test/java/org/apache/cxf/systest/jaxws/Server.java?rev=1310009&r1=1310008&r2=1310009&view=diff
==============================================================================
---
cxf/trunk/systests/jaxws/src/test/java/org/apache/cxf/systest/jaxws/Server.java
(original)
+++
cxf/trunk/systests/jaxws/src/test/java/org/apache/cxf/systest/jaxws/Server.java
Thu Apr 5 18:53:12 2012
@@ -20,15 +20,20 @@
package org.apache.cxf.systest.jaxws;
import java.net.URL;
+import java.util.concurrent.Future;
import javax.jws.WebService;
+import javax.xml.ws.AsyncHandler;
import javax.xml.ws.Endpoint;
+import org.apache.cxf.annotations.UseAsyncMethod;
import org.apache.cxf.jaxws.EndpointImpl;
+import org.apache.cxf.jaxws.ServerAsyncResponse;
import org.apache.cxf.testutil.common.AbstractBusTestServerBase;
import org.apache.hello_world_soap_http.BaseGreeterImpl;
import org.apache.hello_world_soap_http.DocLitBareGreeterImpl;
import org.apache.hello_world_soap_http.GreeterImpl;
+import org.apache.hello_world_soap_http.types.GreetMeResponse;
public class Server extends AbstractBusTestServerBase {
static final String PORT = allocatePort(Server.class);
@@ -42,6 +47,11 @@ public class Server extends AbstractBusT
}
Object implementor;
String address;
+
+ implementor = new AsyncGreeter();
+ address = "http://localhost:"; + PORT + "/SoapContext/AsyncSoapPort";
+ Endpoint.publish(address, implementor);
+
implementor = new GreeterImplMultiPort();
address = "http://localhost:"; + PORT + "/MultiPort/GreeterPort";
Endpoint.publish(address, implementor);
@@ -53,7 +63,7 @@ public class Server extends AbstractBusT
implementor = new GreeterImpl();
address = "http://localhost:"; + PORT + "/SoapContext/SoapPort";
Endpoint.publish(address, implementor);
-
+
implementor = new RefGreeterImpl();
address = "http://localhost:"; + PORT + "/SoapContext/SoapPort2";
Endpoint.publish(address, implementor);
@@ -79,9 +89,38 @@ public class Server extends AbstractBusT
@WebService(endpointInterface = "org.apache.hello_world_soap_http.Greeter",
targetNamespace = "http://apache.org/hello_world_soap_http";)
public class Greeter12Impl extends BaseGreeterImpl {
+
}
-
+ @WebService(endpointInterface = "org.apache.hello_world_soap_http.Greeter",
+ targetNamespace = "http://apache.org/hello_world_soap_http";)
+ public class AsyncGreeter extends BaseGreeterImpl {
+
+ @UseAsyncMethod
+ public String greetMe(String me) {
+ throw new RuntimeException("Should be async");
+ }
+
+ public Future<?> greetMeAsync(final String requestType,
+ final AsyncHandler<GreetMeResponse>
asyncHandler) {
+ final ServerAsyncResponse<GreetMeResponse> r = new
ServerAsyncResponse<GreetMeResponse>();
+ new Thread() {
+ public void run() {
+ try {
+ Thread.sleep(500);
+ } catch (InterruptedException e) {
+ //ignore
+ }
+ GreetMeResponse resp = new GreetMeResponse();
+ resp.setResponseType("Hello " + requestType);
+ r.set(resp);
+ asyncHandler.handleResponse(r);
+ }
+ } .start();
+ return r;
+ }
+ }
+
public static void main(String[] args) {
try {
Server s = new Server();
