Author: sergeyb
Date: Mon Nov 18 22:32:11 2013
New Revision: 1543203
URL: http://svn.apache.org/r1543203
Log:
Merged revisions 1541532 via svnmerge from
https://svn.apache.org/repos/asf/cxf/trunk
........
r1541532 | sergeyb | 2013-11-13 14:21:44 +0000 (Wed, 13 Nov 2013) | 1 line
[CXF-5373] Suspending continuations does not work if the current thread is
different to the one which holds PhaseInterceptorChain lock
........
Modified:
cxf/branches/2.7.x-fixes/ (props changed)
cxf/branches/2.7.x-fixes/api/src/main/java/org/apache/cxf/interceptor/ServiceInvokerInterceptor.java
cxf/branches/2.7.x-fixes/api/src/main/java/org/apache/cxf/message/Message.java
cxf/branches/2.7.x-fixes/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/JAXRSInvoker.java
cxf/branches/2.7.x-fixes/rt/transports/http-jetty/src/main/java/org/apache/cxf/transport/http_jetty/continuations/JettyContinuationWrapper.java
cxf/branches/2.7.x-fixes/rt/transports/http/src/main/java/org/apache/cxf/transport/http/Servlet3ContinuationProvider.java
cxf/branches/2.7.x-fixes/rt/transports/jms/src/main/java/org/apache/cxf/transport/jms/continuations/JMSContinuation.java
cxf/branches/2.7.x-fixes/systests/jaxrs/src/test/java/org/apache/cxf/systest/jaxrs/AbstractJAXRSContinuationsTest.java
cxf/branches/2.7.x-fixes/systests/jaxrs/src/test/java/org/apache/cxf/systest/jaxrs/JAXRSContinuationsServlet3Test.java
cxf/branches/2.7.x-fixes/systests/jaxrs/src/test/resources/jaxrs_async/WEB-INF/beans.xml
Propchange: cxf/branches/2.7.x-fixes/
------------------------------------------------------------------------------
Merged /cxf/trunk:r1541532
Propchange: cxf/branches/2.7.x-fixes/
------------------------------------------------------------------------------
Binary property 'svnmerge-blocked' - no diff available.
Propchange: cxf/branches/2.7.x-fixes/
------------------------------------------------------------------------------
Binary property 'svnmerge-integrated' - no diff available.
Modified:
cxf/branches/2.7.x-fixes/api/src/main/java/org/apache/cxf/interceptor/ServiceInvokerInterceptor.java
URL:
http://svn.apache.org/viewvc/cxf/branches/2.7.x-fixes/api/src/main/java/org/apache/cxf/interceptor/ServiceInvokerInterceptor.java?rev=1543203&r1=1543202&r2=1543203&view=diff
==============================================================================
---
cxf/branches/2.7.x-fixes/api/src/main/java/org/apache/cxf/interceptor/ServiceInvokerInterceptor.java
(original)
+++
cxf/branches/2.7.x-fixes/api/src/main/java/org/apache/cxf/interceptor/ServiceInvokerInterceptor.java
Mon Nov 18 22:32:11 2013
@@ -24,6 +24,7 @@ import java.util.concurrent.ExecutionExc
import java.util.concurrent.Executor;
import java.util.concurrent.FutureTask;
+import org.apache.cxf.common.util.PropertyUtils;
import org.apache.cxf.endpoint.Endpoint;
import org.apache.cxf.message.Exchange;
import org.apache.cxf.message.Message;
@@ -121,7 +122,12 @@ public class ServiceInvokerInterceptor e
} else {
throw new Fault(e.getCause());
}
+ } finally {
+ if
(PropertyUtils.isTrue(exchange.remove(Message.SUSPENDED_INVOCATION))) {
+ message.getInterceptorChain().suspend();
+ }
}
+
}
}
}
Modified:
cxf/branches/2.7.x-fixes/api/src/main/java/org/apache/cxf/message/Message.java
URL:
http://svn.apache.org/viewvc/cxf/branches/2.7.x-fixes/api/src/main/java/org/apache/cxf/message/Message.java?rev=1543203&r1=1543202&r2=1543203&view=diff
==============================================================================
---
cxf/branches/2.7.x-fixes/api/src/main/java/org/apache/cxf/message/Message.java
(original)
+++
cxf/branches/2.7.x-fixes/api/src/main/java/org/apache/cxf/message/Message.java
Mon Nov 18 22:32:11 2013
@@ -139,6 +139,7 @@ public interface Message extends StringM
String WSDL_INTERFACE = "javax.xml.ws.wsdl.interface";
String WSDL_OPERATION = "javax.xml.ws.wsdl.operation";
+ String SUSPENDED_INVOCATION = "org.apache.cxf.suspended.invocation";
/**
* Some properties to allow adding interceptors to the chain
* on a per-request basis. All are a Collection<Interceptor>
Modified:
cxf/branches/2.7.x-fixes/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/JAXRSInvoker.java
URL:
http://svn.apache.org/viewvc/cxf/branches/2.7.x-fixes/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/JAXRSInvoker.java?rev=1543203&r1=1543202&r2=1543203&view=diff
==============================================================================
---
cxf/branches/2.7.x-fixes/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/JAXRSInvoker.java
(original)
+++
cxf/branches/2.7.x-fixes/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/JAXRSInvoker.java
Mon Nov 18 22:32:11 2013
@@ -43,6 +43,7 @@ import org.apache.cxf.common.classloader
import org.apache.cxf.common.i18n.BundleUtils;
import org.apache.cxf.common.logging.LogUtils;
import org.apache.cxf.common.util.ClassHelper;
+import org.apache.cxf.common.util.PropertyUtils;
import org.apache.cxf.helpers.CastUtils;
import org.apache.cxf.interceptor.Fault;
import org.apache.cxf.interceptor.InterceptorChain.State;
@@ -106,7 +107,8 @@ public class JAXRSInvoker extends Abstra
}
return handleFault(ex, exchange.getInMessage());
} finally {
- boolean suspended =
exchange.getInMessage().getInterceptorChain().getState() == State.SUSPENDED;
+ boolean suspended =
PropertyUtils.isTrue(exchange.get(Message.SUSPENDED_INVOCATION))
+ || exchange.getInMessage().getInterceptorChain().getState() ==
State.SUSPENDED;
if (exchange.isOneWay() || suspended) {
ProviderFactory.getInstance(exchange.getInMessage()).clearThreadLocalProxies();
}
Modified:
cxf/branches/2.7.x-fixes/rt/transports/http-jetty/src/main/java/org/apache/cxf/transport/http_jetty/continuations/JettyContinuationWrapper.java
URL:
http://svn.apache.org/viewvc/cxf/branches/2.7.x-fixes/rt/transports/http-jetty/src/main/java/org/apache/cxf/transport/http_jetty/continuations/JettyContinuationWrapper.java?rev=1543203&r1=1543202&r2=1543203&view=diff
==============================================================================
---
cxf/branches/2.7.x-fixes/rt/transports/http-jetty/src/main/java/org/apache/cxf/transport/http_jetty/continuations/JettyContinuationWrapper.java
(original)
+++
cxf/branches/2.7.x-fixes/rt/transports/http-jetty/src/main/java/org/apache/cxf/transport/http_jetty/continuations/JettyContinuationWrapper.java
Mon Nov 18 22:32:11 2013
@@ -25,6 +25,7 @@ import javax.servlet.http.HttpServletRes
import org.apache.cxf.continuations.Continuation;
import org.apache.cxf.continuations.ContinuationCallback;
import org.apache.cxf.message.Message;
+import org.apache.cxf.phase.PhaseInterceptorChain;
import org.apache.cxf.transport.http.AbstractHTTPDestination;
import org.eclipse.jetty.continuation.ContinuationListener;
import org.eclipse.jetty.continuation.ContinuationSupport;
@@ -100,8 +101,13 @@ public class JettyContinuationWrapper im
}
isNew = false;
- // Need to get the right message which is handled in the interceptor
chain
- message.getExchange().getInMessage().getInterceptorChain().suspend();
+ if (PhaseInterceptorChain.getCurrentMessage() == null) {
+ // the current thread is different to the one which holds a lock
on PhaseInterceptorChain
+ message.getExchange().put(Message.SUSPENDED_INVOCATION, true);
+ } else {
+ // Need to get the right message which is handled in the
interceptor chain
+
message.getExchange().getInMessage().getInterceptorChain().suspend();
+ }
continuation.setTimeout(pendingTimeout);
if (!isPending) {
continuation.suspend();
Modified:
cxf/branches/2.7.x-fixes/rt/transports/http/src/main/java/org/apache/cxf/transport/http/Servlet3ContinuationProvider.java
URL:
http://svn.apache.org/viewvc/cxf/branches/2.7.x-fixes/rt/transports/http/src/main/java/org/apache/cxf/transport/http/Servlet3ContinuationProvider.java?rev=1543203&r1=1543202&r2=1543203&view=diff
==============================================================================
---
cxf/branches/2.7.x-fixes/rt/transports/http/src/main/java/org/apache/cxf/transport/http/Servlet3ContinuationProvider.java
(original)
+++
cxf/branches/2.7.x-fixes/rt/transports/http/src/main/java/org/apache/cxf/transport/http/Servlet3ContinuationProvider.java
Mon Nov 18 22:32:11 2013
@@ -31,6 +31,7 @@ import org.apache.cxf.continuations.Cont
import org.apache.cxf.continuations.ContinuationCallback;
import org.apache.cxf.continuations.ContinuationProvider;
import org.apache.cxf.message.Message;
+import org.apache.cxf.phase.PhaseInterceptorChain;
/**
*
@@ -98,9 +99,14 @@ public class Servlet3ContinuationProvide
}
isNew = false;
- // Need to get the right message which is handled in the
interceptor chain
context.setTimeout(timeout);
-
inMessage.getExchange().getInMessage().getInterceptorChain().suspend();
+ if (PhaseInterceptorChain.getCurrentMessage() == null) {
+ // the current thread is different to the one which holds a
lock on PhaseInterceptorChain
+ inMessage.getExchange().put(Message.SUSPENDED_INVOCATION,
true);
+ } else {
+ // Need to get the right message which is handled in the
interceptor chain
+
inMessage.getExchange().getInMessage().getInterceptorChain().suspend();
+ }
return true;
}
Modified:
cxf/branches/2.7.x-fixes/rt/transports/jms/src/main/java/org/apache/cxf/transport/jms/continuations/JMSContinuation.java
URL:
http://svn.apache.org/viewvc/cxf/branches/2.7.x-fixes/rt/transports/jms/src/main/java/org/apache/cxf/transport/jms/continuations/JMSContinuation.java?rev=1543203&r1=1543202&r2=1543203&view=diff
==============================================================================
---
cxf/branches/2.7.x-fixes/rt/transports/jms/src/main/java/org/apache/cxf/transport/jms/continuations/JMSContinuation.java
(original)
+++
cxf/branches/2.7.x-fixes/rt/transports/jms/src/main/java/org/apache/cxf/transport/jms/continuations/JMSContinuation.java
Mon Nov 18 22:32:11 2013
@@ -29,6 +29,7 @@ import org.apache.cxf.common.classloader
import org.apache.cxf.common.logging.LogUtils;
import org.apache.cxf.continuations.Continuation;
import org.apache.cxf.message.Message;
+import org.apache.cxf.phase.PhaseInterceptorChain;
import org.apache.cxf.transport.MessageObserver;
import org.apache.cxf.transport.jms.JMSConfiguration;
import org.apache.cxf.workqueue.WorkQueue;
@@ -138,8 +139,13 @@ public class JMSContinuation implements
if (isPending) {
return false;
}
- // Need to get the right message which is handled in the interceptor
chain
- inMessage.getExchange().getInMessage().getInterceptorChain().suspend();
+ if (PhaseInterceptorChain.getCurrentMessage() == null) {
+ // the current thread is different to the one which holds a lock
on PhaseInterceptorChain
+ inMessage.getExchange().put(Message.SUSPENDED_INVOCATION, true);
+ } else {
+ // Need to get the right message which is handled in the
interceptor chain
+
inMessage.getExchange().getInMessage().getInterceptorChain().suspend();
+ }
updateContinuations(false);
isNew = false;
Modified:
cxf/branches/2.7.x-fixes/systests/jaxrs/src/test/java/org/apache/cxf/systest/jaxrs/AbstractJAXRSContinuationsTest.java
URL:
http://svn.apache.org/viewvc/cxf/branches/2.7.x-fixes/systests/jaxrs/src/test/java/org/apache/cxf/systest/jaxrs/AbstractJAXRSContinuationsTest.java?rev=1543203&r1=1543202&r2=1543203&view=diff
==============================================================================
---
cxf/branches/2.7.x-fixes/systests/jaxrs/src/test/java/org/apache/cxf/systest/jaxrs/AbstractJAXRSContinuationsTest.java
(original)
+++
cxf/branches/2.7.x-fixes/systests/jaxrs/src/test/java/org/apache/cxf/systest/jaxrs/AbstractJAXRSContinuationsTest.java
Mon Nov 18 22:32:11 2013
@@ -43,7 +43,7 @@ public abstract class AbstractJAXRSConti
@Test
public void testDefaultTimeout() throws Exception {
- WebClient wc = WebClient.create("http://localhost:"; + getPort() +
"/bookstore/books/defaulttimeout");
+ WebClient wc = WebClient.create("http://localhost:"; + getPort() +
getBaseAddress() + "/books/defaulttimeout");
WebClient.getConfig(wc).getHttpConduit().getClient().setReceiveTimeout(1000000L);
Response r = wc.get();
assertEquals(503, r.getStatus());
@@ -51,7 +51,7 @@ public abstract class AbstractJAXRSConti
@Test
public void testImmediateResume() throws Exception {
- WebClient wc = WebClient.create("http://localhost:"; + getPort() +
"/bookstore/books/resume");
+ WebClient wc = WebClient.create("http://localhost:"; + getPort() +
getBaseAddress() + "/books/resume");
WebClient.getConfig(wc).getHttpConduit().getClient().setReceiveTimeout(1000000L);
wc.accept("text/plain");
String str = wc.get(String.class);
@@ -61,7 +61,7 @@ public abstract class AbstractJAXRSConti
@Test
public void testImmediateResumeSubresource() throws Exception {
WebClient wc = WebClient.create("http://localhost:"; + getPort()
- +
"/bookstore/books/subresources/books/resume");
+ + getBaseAddress() +
"/books/subresources/books/resume");
WebClient.getConfig(wc).getHttpConduit().getClient().setReceiveTimeout(1000000L);
wc.accept("text/plain");
String str = wc.get(String.class);
@@ -70,7 +70,11 @@ public abstract class AbstractJAXRSConti
@Test
public void testTimeoutAndCancel() throws Exception {
- WebClient wc = WebClient.create("http://localhost:"; + getPort() +
"/bookstore/books/cancel");
+ doTestTimeoutAndCancel(getBaseAddress());
+ }
+
+ protected void doTestTimeoutAndCancel(String baseAddress) throws Exception
{
+ WebClient wc = WebClient.create("http://localhost:"; + getPort() +
baseAddress + "/books/cancel");
WebClient.getConfig(wc).getHttpConduit().getClient().setReceiveTimeout(1000000L);
Response r = wc.get();
assertEquals(503, r.getStatus());
@@ -82,25 +86,25 @@ public abstract class AbstractJAXRSConti
@Test
public void testContinuationWithTimeHandler() throws Exception {
- doTestContinuation("books/timeouthandler");
+ doTestContinuation("/books/timeouthandler");
}
@Test
public void testContinuationWithTimeHandlerResumeOnly() throws Exception {
- doTestContinuation("books/timeouthandlerresume");
+ doTestContinuation("/books/timeouthandlerresume");
}
@Test
public void testContinuation() throws Exception {
- doTestContinuation("books");
+ doTestContinuation("/books");
}
@Test
public void testContinuationSubresource() throws Exception {
- doTestContinuation("books/subresources");
+ doTestContinuation("/books/subresources");
}
protected void doTestContinuation(String pathSegment) throws Exception {
@@ -112,7 +116,7 @@ public abstract class AbstractJAXRSConti
CountDownLatch doneSignal = new CountDownLatch(1);
List<BookWorker> workers = new ArrayList<BookWorker>(5);
for (int x = 1; x < 6; x++) {
- workers.add(new BookWorker("http://localhost:"; + port +
"/bookstore/" + pathSegment + "/" + x,
+ workers.add(new BookWorker("http://localhost:"; + port +
getBaseAddress() + pathSegment + "/" + x,
Integer.toString(x),
"CXF in Action" + x, startSignal,
doneSignal));
}
@@ -187,6 +191,10 @@ public abstract class AbstractJAXRSConti
}
+ protected String getBaseAddress() {
+ return "/bookstore";
+ }
+
protected abstract String getPort();
}
Modified:
cxf/branches/2.7.x-fixes/systests/jaxrs/src/test/java/org/apache/cxf/systest/jaxrs/JAXRSContinuationsServlet3Test.java
URL:
http://svn.apache.org/viewvc/cxf/branches/2.7.x-fixes/systests/jaxrs/src/test/java/org/apache/cxf/systest/jaxrs/JAXRSContinuationsServlet3Test.java?rev=1543203&r1=1543202&r2=1543203&view=diff
==============================================================================
---
cxf/branches/2.7.x-fixes/systests/jaxrs/src/test/java/org/apache/cxf/systest/jaxrs/JAXRSContinuationsServlet3Test.java
(original)
+++
cxf/branches/2.7.x-fixes/systests/jaxrs/src/test/java/org/apache/cxf/systest/jaxrs/JAXRSContinuationsServlet3Test.java
Mon Nov 18 22:32:11 2013
@@ -22,6 +22,7 @@ package org.apache.cxf.systest.jaxrs;
import org.apache.cxf.jaxrs.model.AbstractResourceInfo;
import org.junit.BeforeClass;
+import org.junit.Test;
public class JAXRSContinuationsServlet3Test extends
AbstractJAXRSContinuationsTest {
@@ -36,6 +37,15 @@ public class JAXRSContinuationsServlet3T
}
+ @Test
+ public void testTimeoutAndCancelAsyncExecutor() throws Exception {
+ doTestTimeoutAndCancel("/asyncexecutor/bookstore");
+ }
+
+ protected String getBaseAddress() {
+ return "/async/bookstore";
+ }
+
protected String getPort() {
return PORT;
}
Modified:
cxf/branches/2.7.x-fixes/systests/jaxrs/src/test/resources/jaxrs_async/WEB-INF/beans.xml
URL:
http://svn.apache.org/viewvc/cxf/branches/2.7.x-fixes/systests/jaxrs/src/test/resources/jaxrs_async/WEB-INF/beans.xml?rev=1543203&r1=1543202&r2=1543203&view=diff
==============================================================================
---
cxf/branches/2.7.x-fixes/systests/jaxrs/src/test/resources/jaxrs_async/WEB-INF/beans.xml
(original)
+++
cxf/branches/2.7.x-fixes/systests/jaxrs/src/test/resources/jaxrs_async/WEB-INF/beans.xml
Mon Nov 18 22:32:11 2013
@@ -17,37 +17,38 @@
specific language governing permissions and limitations
under the License.
-->
-<!-- START SNIPPET: beans -->
-<!--beans xmlns="http://www.springframework.org/schema/beans";
- xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance";
- xmlns:simple="http://cxf.apache.org/simple";
- xsi:schemaLocation="
- http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans.xsd
- http://cxf.apache.org/simple http://cxf.apache.org/schemas/simple.xsd"-->
-<beans xmlns="http://www.springframework.org/schema/beans";
- xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance";
- xmlns:util="http://www.springframework.org/schema/util";
- xmlns:jaxrs="http://cxf.apache.org/jaxrs";
- xmlns:cxf="http://cxf.apache.org/core";
- xsi:schemaLocation="
-http://www.springframework.org/schema/beans
-http://www.springframework.org/schema/beans/spring-beans.xsd
-http://www.springframework.org/schema/util
-http://www.springframework.org/schema/util/spring-util.xsd
-http://cxf.apache.org/jaxrs
-http://cxf.apache.org/schemas/jaxrs.xsd
-http://cxf.apache.org/core
-http://cxf.apache.org/schemas/core.xsd";>
-
- <import resource="classpath:META-INF/cxf/cxf.xml" />
- <import resource="classpath:META-INF/cxf/cxf-servlet.xml" />
-
- <bean class="org.apache.cxf.systest.jaxrs.BookContinuationStore"
id="serviceBean"/>
-
- <jaxrs:server id="bookservice" address="/">
- <jaxrs:serviceBeans>
- <ref bean="serviceBean" />
- </jaxrs:serviceBeans>
- </jaxrs:server>
+<beans xmlns="http://www.springframework.org/schema/beans";
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance";
+ xmlns:jaxrs="http://cxf.apache.org/jaxrs";
+ xmlns:task="http://www.springframework.org/schema/task";
+ xsi:schemaLocation="
+ http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans.xsd
+ http://www.springframework.org/schema/task
http://www.springframework.org/schema/task/spring-task.xsd
+ http://cxf.apache.org/jaxrs http://cxf.apache.org/schemas/jaxrs.xsd";>
+
+ <import resource="classpath:META-INF/cxf/cxf.xml"/>
+ <import resource="classpath:META-INF/cxf/cxf-servlet.xml"/>
+
+ <bean class="org.apache.cxf.systest.jaxrs.BookContinuationStore"
id="serviceBean"/>
+
+ <jaxrs:server id="bookservice" address="/async">
+ <jaxrs:serviceBeans>
+ <ref bean="serviceBean"/>
+ </jaxrs:serviceBeans>
+ </jaxrs:server>
+
+ <jaxrs:server id="bookservice2" address="/asyncexecutor">
+
+ <jaxrs:executor>
+ <ref bean="workerPool" />
+ </jaxrs:executor>
+
+ <jaxrs:serviceBeans>
+ <ref bean="serviceBean"/>
+ </jaxrs:serviceBeans>
+ </jaxrs:server>
+
+ <task:executor id="workerPool" pool-size="1" queue-capacity="1"/>
+
</beans>
<!-- END SNIPPET: beans -->
