Implement asynchronous callbacks for CometEvent.close and CometEvent.setTimeout to...

CometProcessor implements servlet interface, or the code will fail deployment
if no keepalive, close the connection after comet transaction is complete

git-svn-id: https://svn.apache.org/repos/asf/tomcat/trunk@645175 13f79535-47bb-0310-9956-ffa450edef68

diff --git a/java/org/apache/catalina/CometProcessor.java b/java/org/apache/catalina/CometProcessor.java
index 8645f00..4536a82 100644 (file)
--- a/java/org/apache/catalina/CometProcessor.java
+++ b/java/org/apache/catalina/CometProcessor.java
@@ -21,6 +21,7 @@ package org.apache.catalina;
 import java.io.IOException;
 
 import javax.servlet.ServletException;
+import javax.servlet.Servlet;
 
 /**
  * This interface should be implemented by servlets which would like to handle
@@ -29,7 +30,7 @@ import javax.servlet.ServletException;
  * Note: When this interface is implemented, the service method of the servlet will
  * never be called, and will be replaced with a begin event.
  */
-public interface CometProcessor {
+public interface CometProcessor extends Servlet{
 
     /**
      * Process the given Comet event.

diff --git a/java/org/apache/catalina/connector/CometEventImpl.java b/java/org/apache/catalina/connector/CometEventImpl.java
index 5a2670e..3c3489a 100644 (file)
--- a/java/org/apache/catalina/connector/CometEventImpl.java
+++ b/java/org/apache/catalina/connector/CometEventImpl.java
@@ -26,6 +26,7 @@ import javax.servlet.http.HttpServletResponse;
 
 import org.apache.catalina.CometEvent;
 import org.apache.catalina.util.StringManager;
+import org.apache.coyote.ActionCode;
 
 public class CometEventImpl implements CometEvent {
 
@@ -92,8 +93,10 @@ public class CometEventImpl implements CometEvent {
         if (request == null) {
             throw new IllegalStateException(sm.getString("cometEvent.nullRequest"));
         }
+        boolean iscomet = request.isComet();
         request.setComet(false);
         response.finishResponse();
+        if (iscomet) request.cometClose();
     }
 
     public EventSubType getEventSubType() {
@@ -116,6 +119,7 @@ public class CometEventImpl implements CometEvent {
             UnsupportedOperationException {
         if (request.getAttribute("org.apache.tomcat.comet.timeout.support") == Boolean.TRUE) {
             request.setAttribute("org.apache.tomcat.comet.timeout", new Integer(timeout));
+            if (request.isComet()) request.setCometTimeout((long)timeout);
         } else {
             throw new UnsupportedOperationException();
         }

diff --git a/java/org/apache/catalina/connector/Request.java b/java/org/apache/catalina/connector/Request.java
index 7a2f21f..2b3ed3f 100644 (file)
--- a/java/org/apache/catalina/connector/Request.java
+++ b/java/org/apache/catalina/connector/Request.java
@@ -2256,6 +2256,13 @@ public class Request
         return (inputBuffer.available() > 0);
     }
 
+    public void cometClose() {
+        coyoteRequest.action(ActionCode.ACTION_COMET_CLOSE,getEvent());
+    }
+    
+    public void setCometTimeout(long timeout) {
+        coyoteRequest.action(ActionCode.ACTION_COMET_SETTIMEOUT,new Long(timeout));
+    }
     
     // ------------------------------------------------------ Protected Methods
 

diff --git a/java/org/apache/coyote/ActionCode.java b/java/org/apache/coyote/ActionCode.java
index 1bc8622..ac2e0fc 100644 (file)
--- a/java/org/apache/coyote/ActionCode.java
+++ b/java/org/apache/coyote/ActionCode.java
@@ -141,7 +141,7 @@ public final class ActionCode {
 
 
     /**
-     * Callback for begin Comet processing
+     * Callback for end Comet processing
      */
     public static final ActionCode ACTION_COMET_END = new ActionCode(22);
 
@@ -151,7 +151,16 @@ public final class ActionCode {
      */
     public static final ActionCode ACTION_AVAILABLE = new ActionCode(23);
 
+    /**
+     * Callback for an asynchronous close of the Comet event
+     */
+    public static final ActionCode ACTION_COMET_CLOSE = new ActionCode(24);
 
+    /**
+     * Callback for setting the timeout asynchronously
+     */
+    public static final ActionCode ACTION_COMET_SETTIMEOUT = new ActionCode(25);
+    
     // ----------------------------------------------------------- Constructors
     int code;
 

diff --git a/java/org/apache/coyote/http11/Http11AprProcessor.java b/java/org/apache/coyote/http11/Http11AprProcessor.java
index 884b71c..d1a8ead 100644 (file)
--- a/java/org/apache/coyote/http11/Http11AprProcessor.java
+++ b/java/org/apache/coyote/http11/Http11AprProcessor.java
@@ -1205,6 +1205,10 @@ public class Http11AprProcessor implements ActionHook {
             comet = true;
         } else if (actionCode == ActionCode.ACTION_COMET_END) {
             comet = false;
+        } else if (actionCode == ActionCode.ACTION_COMET_CLOSE) {
+            //no op
+        } else if (actionCode == ActionCode.ACTION_COMET_SETTIMEOUT) {
+            //no op
         }
 
     }

diff --git a/java/org/apache/coyote/http11/Http11NioProcessor.java b/java/org/apache/coyote/http11/Http11NioProcessor.java
index 0c57c7b..86df02a 100644 (file)
--- a/java/org/apache/coyote/http11/Http11NioProcessor.java
+++ b/java/org/apache/coyote/http11/Http11NioProcessor.java
@@ -777,7 +777,8 @@ public class Http11NioProcessor implements ActionHook {
             return SocketState.CLOSED;
         } else if (!comet) {
             recycle();
-            return SocketState.OPEN;
+            //pay attention to the keep alive flag set in process()
+            return (keepAlive)?SocketState.OPEN:SocketState.CLOSED;
         } else {
             return SocketState.LONG;
         }
@@ -1219,6 +1220,21 @@ public class Http11NioProcessor implements ActionHook {
             comet = true;
         } else if (actionCode == ActionCode.ACTION_COMET_END) {
             comet = false;
+        }  else if (actionCode == ActionCode.ACTION_COMET_CLOSE) {
+            NioEndpoint.KeyAttachment attach = (NioEndpoint.KeyAttachment)socket.getAttachment(false);
+            attach.setCometOps(NioEndpoint.OP_CALLBACK);
+            //notify poller if not on a tomcat thread
+            RequestInfo rp = request.getRequestProcessor();
+            if ( rp.getStage() != org.apache.coyote.Constants.STAGE_SERVICE ) //async handling
+                socket.getPoller().cometInterest(socket);
+        } else if (actionCode == ActionCode.ACTION_COMET_SETTIMEOUT) {
+            if (param==null) return;
+            NioEndpoint.KeyAttachment attach = (NioEndpoint.KeyAttachment)socket.getAttachment(false);
+            long timeout = ((Long)param).longValue();
+            //if we are not piggy backing on a worker thread, set the timeout
+            RequestInfo rp = request.getRequestProcessor();
+            if ( rp.getStage() != org.apache.coyote.Constants.STAGE_SERVICE ) //async handling
+                attach.setTimeout(timeout);
         }
 
     }

diff --git a/java/org/apache/tomcat/util/net/NioEndpoint.java b/java/org/apache/tomcat/util/net/NioEndpoint.java
index bd91666..55eee32 100644 (file)
--- a/java/org/apache/tomcat/util/net/NioEndpoint.java
+++ b/java/org/apache/tomcat/util/net/NioEndpoint.java
@@ -1635,6 +1635,7 @@ public class NioEndpoint {
         protected void reg(SelectionKey sk, KeyAttachment attachment, int intops) {
             sk.interestOps(intops); 
             attachment.interestOps(intops);
+            attachment.setCometOps(intops);
         }
 
         protected void timeout(int keyCount, boolean hasEvents) {
@@ -1659,6 +1660,7 @@ public class NioEndpoint {
                     } else if ( ka.getError() ) {
                         cancelledKey(key, SocketStatus.ERROR,true);
                     } else if (ka.getComet() && ka.getCometNotify() ) {
+                        ka.setCometNotify(false);
                         reg(key,ka,0);//avoid multiple calls, this gets reregistered after invokation
                         //if (!processSocket(ka.getChannel(), SocketStatus.OPEN_CALLBACK)) processSocket(ka.getChannel(), SocketStatus.DISCONNECT);
                         if (!processSocket(ka.getChannel(), SocketStatus.OPEN)) processSocket(ka.getChannel(), SocketStatus.DISCONNECT);