*/
protected NioEndpoint.SendfileData sendfileData = null;
- /**
- * Closed flag, a Comet async thread can
- * signal for this Nio processor to be closed and recycled instead
- * of waiting for a timeout.
- * Closed by HttpServletResponse.getWriter().close()
- */
- protected boolean cometClose = false;
-
+
/**
* Socket associated with the current connection.
*/
@Override
public void recycleInternal() {
socket = null;
- cometClose = false;
comet = false;
sendfileData = null;
}
@Override
public void actionInternal(ActionCode actionCode, Object param) {
- if (actionCode == ActionCode.CLOSE) {
- // Close
- // End the processing of the current request, and stop any further
- // transactions with the client
-
- comet = false;
- cometClose = true;
-
- try {
- outputBuffer.endRequest();
- } catch (IOException e) {
- // Set error flag
- error = true;
- }
-
- } else if (actionCode == ActionCode.REQ_HOST_ADDR_ATTRIBUTE) {
+ if (actionCode == ActionCode.REQ_HOST_ADDR_ATTRIBUTE) {
// Get remote host address
if ((remoteAddr == null) && (socket != null)) {
if (socket==null || socket.getSocket().getAttachment(false)==null) return;
NioEndpoint.KeyAttachment attach = (NioEndpoint.KeyAttachment)socket.getSocket().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
+ if (rp.getStage() != org.apache.coyote.Constants.STAGE_SERVICE) {
+ // Close event for this processor triggered by request
+ // processing in another processor, a non-Tomcat thread (i.e.
+ // an application controlled thread) or similar.
socket.getSocket().getPoller().add(socket.getSocket());
+ }
} else if (actionCode == ActionCode.COMET_SETTIMEOUT) {
if (param==null) return;
if (socket==null || socket.getSocket().getAttachment(false)==null) return;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
import org.junit.Test;
import org.apache.catalina.Context;
import org.apache.catalina.comet.CometEvent.EventType;
+import org.apache.catalina.connector.CometEventImpl;
+import org.apache.catalina.connector.Request;
+import org.apache.catalina.connector.Response;
import org.apache.catalina.startup.Tomcat;
import org.apache.catalina.startup.TomcatBaseTest;
+import org.apache.catalina.valves.ValveBase;
public class TestCometProcessor extends TomcatBaseTest {
@Test
+ public void testAsyncClose() throws Exception {
+
+ if (!isCometSupported()) {
+ return;
+ }
+
+ // Setup Tomcat instance
+ Tomcat tomcat = getTomcatInstance();
+ Context root = tomcat.addContext("", TEMP_DIR);
+ Tomcat.addServlet(root, "comet", new SimpleCometServlet());
+ root.addServletMapping("/comet", "comet");
+ Tomcat.addServlet(root, "hello", new HelloWorldServlet());
+ root.addServletMapping("/hello", "hello");
+ root.getPipeline().addValve(new AsyncCometCloseValve());
+ tomcat.getConnector().setProperty("connectionTimeout", "5000");
+ tomcat.start();
+
+ // Create connection to Comet servlet
+ final Socket socket =
+ SocketFactory.getDefault().createSocket("localhost", getPort());
+ socket.setSoTimeout(5000);
+
+ final OutputStream os = socket.getOutputStream();
+ String requestLine = "POST http://localhost:" + getPort() +
+ "/comet HTTP/1.1\r\n";
+ os.write(requestLine.getBytes());
+ os.write("transfer-encoding: chunked\r\n".getBytes());
+ os.write("\r\n".getBytes());
+
+ InputStream is = socket.getInputStream();
+ ResponseReaderThread readThread = new ResponseReaderThread(is);
+ readThread.start();
+
+ // Wait for the comet request/response to finish
+ int count = 0;
+ while (count < 10 && !readThread.getResponse().endsWith("0\r\n\r\n")) {
+ Thread.sleep(500);
+ count++;
+ }
+
+ if (count == 10) {
+ fail("Comet request did not complete");
+ }
+
+ // Send a standard HTTP request on the same connection
+ requestLine = "GET http://localhost:" + getPort() +
+ "/hello HTTP/1.1\r\n";
+ os.write(requestLine.getBytes());
+ os.write("\r\n".getBytes());
+
+ // Check for the expected response
+ count = 0;
+ while (count < 10 && !readThread.getResponse().contains(
+ HelloWorldServlet.RESPONSE_TEXT)) {
+ Thread.sleep(500);
+ count++;
+ }
+
+ if (count == 10) {
+ fail("Non-comet request did not complete");
+ }
+
+ readThread.join();
+ os.close();
+ is.close();
+ }
+
+ @Test
public void testSimpleCometClient() throws Exception {
if (!isCometSupported()) {
}
}
}
+
+ private static class AsyncCometCloseValve extends ValveBase {
+
+ @Override
+ public void invoke(Request request, Response response)
+ throws IOException, ServletException {
+
+ CometEventImpl event = new CometEventImpl(request, response);
+
+ getNext().invoke(request, response);
+
+ if (request.isComet()) {
+ Thread t = new AsyncCometCloseThread(event);
+ t.start();
+ }
+ }
+ }
+
+ private static class AsyncCometCloseThread extends Thread {
+
+ private CometEvent event;
+
+ public AsyncCometCloseThread(CometEvent event) {
+ this.event = event;
+ }
+
+ @Override
+ public void run() {
+ try {
+ Thread.sleep(2000);
+ event.close();
+ } catch (Exception e) {
+ // Test should fail. Report what went wrong.
+ e.printStackTrace();
+ }
+ }
+ }
}
projects / tomcat7.0 / commitdiff