Author: dkulp
Date: Fri Nov 22 17:28:11 2013
New Revision: 1544608
URL: http://svn.apache.org/r1544608
Log:
[CXF-5414] If a continuation is complete, make sure we let he container know so
timeouts will cancel.
Modified:
cxf/trunk/core/src/main/java/org/apache/cxf/continuations/ContinuationProvider.java
cxf/trunk/rt/transports/http-jetty/src/main/java/org/apache/cxf/transport/http_jetty/JettyHTTPDestination.java
cxf/trunk/rt/transports/http-jetty/src/main/java/org/apache/cxf/transport/http_jetty/continuations/JettyContinuationProvider.java
cxf/trunk/rt/transports/http-netty/netty-server/src/main/java/org/apache/cxf/transport/http/netty/server/NettyHttpDestination.java
cxf/trunk/rt/transports/http/src/main/java/org/apache/cxf/transport/http/AbstractHTTPDestination.java
cxf/trunk/rt/transports/http/src/main/java/org/apache/cxf/transport/http/Servlet3ContinuationProvider.java
cxf/trunk/rt/transports/jms/src/main/java/org/apache/cxf/transport/jms/continuations/JMSContinuationProvider.java
Modified:
cxf/trunk/core/src/main/java/org/apache/cxf/continuations/ContinuationProvider.java
URL:
http://svn.apache.org/viewvc/cxf/trunk/core/src/main/java/org/apache/cxf/continuations/ContinuationProvider.java?rev=1544608&r1=1544607&r2=1544608&view=diff
==============================================================================
---
cxf/trunk/core/src/main/java/org/apache/cxf/continuations/ContinuationProvider.java
(original)
+++
cxf/trunk/core/src/main/java/org/apache/cxf/continuations/ContinuationProvider.java
Fri Nov 22 17:28:11 2013
@@ -30,4 +30,9 @@ public interface ContinuationProvider {
* @return transport-neutral ContinuationWrapper
*/
Continuation getContinuation();
+
+ /**
+ * Let the provider know that the exchange is complete
+ */
+ void complete();
}
Modified:
cxf/trunk/rt/transports/http-jetty/src/main/java/org/apache/cxf/transport/http_jetty/JettyHTTPDestination.java
URL:
http://svn.apache.org/viewvc/cxf/trunk/rt/transports/http-jetty/src/main/java/org/apache/cxf/transport/http_jetty/JettyHTTPDestination.java?rev=1544608&r1=1544607&r2=1544608&view=diff
==============================================================================
---
cxf/trunk/rt/transports/http-jetty/src/main/java/org/apache/cxf/transport/http_jetty/JettyHTTPDestination.java
(original)
+++
cxf/trunk/rt/transports/http-jetty/src/main/java/org/apache/cxf/transport/http_jetty/JettyHTTPDestination.java
Fri Nov 22 17:28:11 2013
@@ -287,6 +287,10 @@ public class JettyHTTPDestination extend
incomingObserver.onMessage(inMessage);
resp.flushBuffer();
baseRequest.setHandled(true);
+ ContinuationProvider p = inMessage.get(ContinuationProvider.class);
+ if (p != null) {
+ p.complete();
+ }
} catch (SuspendedInvocationException ex) {
if (ex.getRuntimeException() != null) {
throw ex.getRuntimeException();
Modified:
cxf/trunk/rt/transports/http-jetty/src/main/java/org/apache/cxf/transport/http_jetty/continuations/JettyContinuationProvider.java
URL:
http://svn.apache.org/viewvc/cxf/trunk/rt/transports/http-jetty/src/main/java/org/apache/cxf/transport/http_jetty/continuations/JettyContinuationProvider.java?rev=1544608&r1=1544607&r2=1544608&view=diff
==============================================================================
---
cxf/trunk/rt/transports/http-jetty/src/main/java/org/apache/cxf/transport/http_jetty/continuations/JettyContinuationProvider.java
(original)
+++
cxf/trunk/rt/transports/http-jetty/src/main/java/org/apache/cxf/transport/http_jetty/continuations/JettyContinuationProvider.java
Fri Nov 22 17:28:11 2013
@@ -40,6 +40,14 @@ public class JettyContinuationProvider i
response = resp;
this.inMessage = m;
}
+
+ public void complete() {
+ JettyContinuationWrapper r = getContinuation(false);
+ if (r != null) {
+ r.reset();
+ }
+ wrapper = null;
+ }
public Continuation getContinuation() {
return getContinuation(true);
}
Modified:
cxf/trunk/rt/transports/http-netty/netty-server/src/main/java/org/apache/cxf/transport/http/netty/server/NettyHttpDestination.java
URL:
http://svn.apache.org/viewvc/cxf/trunk/rt/transports/http-netty/netty-server/src/main/java/org/apache/cxf/transport/http/netty/server/NettyHttpDestination.java?rev=1544608&r1=1544607&r2=1544608&view=diff
==============================================================================
---
cxf/trunk/rt/transports/http-netty/netty-server/src/main/java/org/apache/cxf/transport/http/netty/server/NettyHttpDestination.java
(original)
+++
cxf/trunk/rt/transports/http-netty/netty-server/src/main/java/org/apache/cxf/transport/http/netty/server/NettyHttpDestination.java
Fri Nov 22 17:28:11 2013
@@ -25,15 +25,18 @@ import java.net.MalformedURLException;
import java.net.URL;
import java.util.logging.Level;
import java.util.logging.Logger;
+
import javax.servlet.ServletContext;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
+
import org.apache.cxf.Bus;
import org.apache.cxf.BusFactory;
import org.apache.cxf.common.classloader.ClassLoaderUtils;
import org.apache.cxf.common.classloader.ClassLoaderUtils.ClassLoaderHolder;
import org.apache.cxf.common.logging.LogUtils;
import org.apache.cxf.common.util.StringUtils;
+import org.apache.cxf.continuations.ContinuationProvider;
import org.apache.cxf.continuations.SuspendedInvocationException;
import org.apache.cxf.interceptor.Fault;
import org.apache.cxf.message.ExchangeImpl;
@@ -208,7 +211,12 @@ public class NettyHttpDestination extend
try {
incomingObserver.onMessage(inMessage);
+
resp.flushBuffer();
+ ContinuationProvider p = inMessage.get(ContinuationProvider.class);
+ if (p != null) {
+ p.complete();
+ }
} catch (SuspendedInvocationException ex) {
if (ex.getRuntimeException() != null) {
throw ex.getRuntimeException();
Modified:
cxf/trunk/rt/transports/http/src/main/java/org/apache/cxf/transport/http/AbstractHTTPDestination.java
URL:
http://svn.apache.org/viewvc/cxf/trunk/rt/transports/http/src/main/java/org/apache/cxf/transport/http/AbstractHTTPDestination.java?rev=1544608&r1=1544607&r2=1544608&view=diff
==============================================================================
---
cxf/trunk/rt/transports/http/src/main/java/org/apache/cxf/transport/http/AbstractHTTPDestination.java
(original)
+++
cxf/trunk/rt/transports/http/src/main/java/org/apache/cxf/transport/http/AbstractHTTPDestination.java
Fri Nov 22 17:28:11 2013
@@ -238,6 +238,10 @@ public abstract class AbstractHTTPDestin
try {
incomingObserver.onMessage(inMessage);
+ ContinuationProvider p = inMessage.get(ContinuationProvider.class);
+ if (p != null) {
+ p.complete();
+ }
} catch (SuspendedInvocationException ex) {
if (ex.getRuntimeException() != null) {
throw ex.getRuntimeException();
Modified:
cxf/trunk/rt/transports/http/src/main/java/org/apache/cxf/transport/http/Servlet3ContinuationProvider.java
URL:
http://svn.apache.org/viewvc/cxf/trunk/rt/transports/http/src/main/java/org/apache/cxf/transport/http/Servlet3ContinuationProvider.java?rev=1544608&r1=1544607&r2=1544608&view=diff
==============================================================================
---
cxf/trunk/rt/transports/http/src/main/java/org/apache/cxf/transport/http/Servlet3ContinuationProvider.java
(original)
+++
cxf/trunk/rt/transports/http/src/main/java/org/apache/cxf/transport/http/Servlet3ContinuationProvider.java
Fri Nov 22 17:28:11 2013
@@ -48,6 +48,14 @@ public class Servlet3ContinuationProvide
this.req = req;
this.resp = resp;
}
+
+ public void complete() {
+ if (continuation != null) {
+ continuation.reset();
+ continuation = null;
+ }
+ }
+
/** {@inheritDoc}*/
public Continuation getContinuation() {
@@ -77,7 +85,7 @@ public class Servlet3ContinuationProvide
context = req.startAsync(req, resp);
context.addListener(this);
}
-
+
void startAsyncAgain() {
AsyncContext old = context;
Modified:
cxf/trunk/rt/transports/jms/src/main/java/org/apache/cxf/transport/jms/continuations/JMSContinuationProvider.java
URL:
http://svn.apache.org/viewvc/cxf/trunk/rt/transports/jms/src/main/java/org/apache/cxf/transport/jms/continuations/JMSContinuationProvider.java?rev=1544608&r1=1544607&r2=1544608&view=diff
==============================================================================
---
cxf/trunk/rt/transports/jms/src/main/java/org/apache/cxf/transport/jms/continuations/JMSContinuationProvider.java
(original)
+++
cxf/trunk/rt/transports/jms/src/main/java/org/apache/cxf/transport/jms/continuations/JMSContinuationProvider.java
Fri Nov 22 17:28:11 2013
@@ -51,6 +51,12 @@ public class JMSContinuationProvider imp
this.jmsListener = jmsListener;
this.jmsConfig = jmsConfig;
}
+ public void complete() {
+ JMSContinuation cw = inMessage.get(JMSContinuation.class);
+ if (cw != null) {
+ cw.reset();
+ }
+ }
public Continuation getContinuation() {
Message m = inMessage;
