Author: remy.maucherat(a)jboss.com Date: 2013-03-20 13:30:48 -0400 (Wed, 20 Mar 2013) New Revision: 2177 Modified: branches/8.0.x/src/main/java/org/apache/catalina/connector/CoyoteInputStream.java branches/8.0.x/src/main/java/org/apache/catalina/connector/CoyoteOutputStream.java branches/8.0.x/src/main/java/org/apache/catalina/connector/HttpEventImpl.java branches/8.0.x/src/main/java/org/apache/catalina/connector/InputBuffer.java branches/8.0.x/src/main/java/org/apache/catalina/connector/OutputBuffer.java branches/8.0.x/src/main/java/org/apache/catalina/connector/Request.java branches/8.0.x/src/main/java/org/apache/catalina/connector/Response.java branches/8.0.x/src/main/java/org/apache/catalina/core/StandardWrapperValve.java branches/8.0.x/src/main/java/org/apache/coyote/ActionCode.java branches/8.0.x/src/main/java/org/apache/coyote/http11/Http11AprProcessor.java branches/8.0.x/src/main/java/org/apache/coyote/http11/Http11NioProcessor.java branches/8.0.x/src/main/java/org/jboss/web/CatalinaMessages.java Log: Start experimenting with 3.1. Modified: branches/8.0.x/src/main/java/org/apache/catalina/connector/CoyoteInputStream.java =================================================================== --- branches/8.0.x/src/main/java/org/apache/catalina/connector/CoyoteInputStream.java 2013-03-18 16:16:55 UTC (rev 2176) +++ branches/8.0.x/src/main/java/org/apache/catalina/connector/CoyoteInputStream.java 2013-03-20 17:30:48 UTC (rev 2177) @@ -243,8 +243,8 @@ @Override - public void setReadListener(ReadListener arg0) { - // TODO Auto-generated method stub + public void setReadListener(ReadListener readListener) { + ib.setReadListener(readListener); } } Modified: branches/8.0.x/src/main/java/org/apache/catalina/connector/CoyoteOutputStream.java =================================================================== --- branches/8.0.x/src/main/java/org/apache/catalina/connector/CoyoteOutputStream.java 2013-03-18 16:16:55 UTC (rev 2176) +++ branches/8.0.x/src/main/java/org/apache/catalina/connector/CoyoteOutputStream.java 2013-03-20 17:30:48 UTC (rev 2177) @@ -112,9 +112,8 @@ @Override - public void setWriteListener(WriteListener arg0) { - // TODO Auto-generated method stub - + public void setWriteListener(WriteListener writeListener) { + ob.setWriteListener(writeListener); } Modified: branches/8.0.x/src/main/java/org/apache/catalina/connector/HttpEventImpl.java =================================================================== --- branches/8.0.x/src/main/java/org/apache/catalina/connector/HttpEventImpl.java 2013-03-18 16:16:55 UTC (rev 2176) +++ branches/8.0.x/src/main/java/org/apache/catalina/connector/HttpEventImpl.java 2013-03-20 17:30:48 UTC (rev 2177) @@ -20,12 +20,15 @@ import java.io.IOException; +import javax.servlet.ServletInputStream; +import javax.servlet.ServletOutputStream; import javax.servlet.http.HttpServletRequest; import javax.servlet.http.HttpServletResponse; +import javax.servlet.http.WebConnection; import org.jboss.servlet.http.HttpEvent; -public class HttpEventImpl implements HttpEvent { +public class HttpEventImpl implements HttpEvent, WebConnection { public HttpEventImpl(Request request, Response response) { this.request = request; @@ -85,6 +88,14 @@ return response.getResponse(); } + public ServletInputStream getInputStream() throws IOException { + return request.getInputStream(); + } + + public ServletOutputStream getOutputStream() throws IOException { + return response.getOutputStream(); + } + public void setTimeout(int timeout) { request.setTimeout(timeout); } Modified: branches/8.0.x/src/main/java/org/apache/catalina/connector/InputBuffer.java =================================================================== --- branches/8.0.x/src/main/java/org/apache/catalina/connector/InputBuffer.java 2013-03-18 16:16:55 UTC (rev 2176) +++ branches/8.0.x/src/main/java/org/apache/catalina/connector/InputBuffer.java 2013-03-20 17:30:48 UTC (rev 2177) @@ -27,6 +27,8 @@ import java.util.HashMap; import java.util.Locale; +import javax.servlet.ReadListener; + import org.apache.catalina.security.SecurityUtil; import org.apache.coyote.ActionCode; import org.apache.coyote.Request; @@ -143,6 +145,12 @@ private int size = -1; + /** + * Read listener. + */ + private ReadListener readListener = null; + + // ----------------------------------------------------------- Constructors @@ -229,6 +237,7 @@ gotEnc = false; enc = null; + readListener = null; } @@ -575,4 +584,16 @@ } + public ReadListener getReadListener() { + return readListener; + } + + public void setReadListener(ReadListener readListener) { + if (this.readListener != null) { + throw MESSAGES.readListenerAlreadySet(); + } + this.readListener = readListener; + coyoteRequest.action(ActionCode.ACTION_EVENT_READ_BEGIN, null); + } + } Modified: branches/8.0.x/src/main/java/org/apache/catalina/connector/OutputBuffer.java =================================================================== --- branches/8.0.x/src/main/java/org/apache/catalina/connector/OutputBuffer.java 2013-03-18 16:16:55 UTC (rev 2176) +++ branches/8.0.x/src/main/java/org/apache/catalina/connector/OutputBuffer.java 2013-03-20 17:30:48 UTC (rev 2177) @@ -18,6 +18,8 @@ package org.apache.catalina.connector; +import static org.jboss.web.CatalinaMessages.MESSAGES; + import java.io.IOException; import java.io.Writer; import java.security.AccessController; @@ -26,6 +28,8 @@ import java.util.HashMap; import java.util.Locale; +import javax.servlet.WriteListener; + import org.apache.coyote.ActionCode; import org.apache.coyote.Response; import org.apache.catalina.Globals; @@ -148,6 +152,12 @@ private boolean suspended = false; + /** + * Write listener. + */ + private WriteListener writeListener = null; + + // ----------------------------------------------------------- Constructors @@ -257,6 +267,7 @@ gotEnc = false; enc = null; + writeListener = null; } @@ -662,5 +673,16 @@ return bb.getLimit(); } + public WriteListener getWriteListener() { + return writeListener; + } + public void setWriteListener(WriteListener writeListener) { + if (this.writeListener != null) { + throw MESSAGES.writeListenerAlreadySet(); + } + this.writeListener = writeListener; + coyoteResponse.action(ActionCode.ACTION_EVENT_WRITE_BEGIN, null); + } + } Modified: branches/8.0.x/src/main/java/org/apache/catalina/connector/Request.java =================================================================== --- branches/8.0.x/src/main/java/org/apache/catalina/connector/Request.java 2013-03-18 16:16:55 UTC (rev 2176) +++ branches/8.0.x/src/main/java/org/apache/catalina/connector/Request.java 2013-03-20 17:30:48 UTC (rev 2177) @@ -49,6 +49,7 @@ import javax.servlet.AsyncEvent; import javax.servlet.AsyncListener; import javax.servlet.DispatcherType; +import javax.servlet.ReadListener; import javax.servlet.RequestDispatcher; import javax.servlet.ServletContext; import javax.servlet.ServletException; @@ -94,9 +95,9 @@ import org.apache.tomcat.util.http.Parameters; import org.apache.tomcat.util.http.ServerCookie; import org.apache.tomcat.util.http.fileupload.FileItem; -import org.apache.tomcat.util.http.fileupload.FileUploadException; import org.apache.tomcat.util.http.fileupload.FileUploadBase.FileSizeLimitExceededException; import org.apache.tomcat.util.http.fileupload.FileUploadBase.SizeLimitExceededException; +import org.apache.tomcat.util.http.fileupload.FileUploadException; import org.apache.tomcat.util.http.fileupload.disk.DiskFileItemFactory; import org.apache.tomcat.util.http.fileupload.servlet.ServletFileUpload; import org.apache.tomcat.util.http.mapper.MappingData; @@ -488,6 +489,12 @@ protected LinkedList<AsyncListener> asyncListenerInstances = new LinkedList<AsyncListener>(); + /** + * Upgrade handler. + */ + protected HttpUpgradeHandler upgradeHandler = null; + + // --------------------------------------------------------- Public Methods @@ -519,6 +526,7 @@ event.clear(); event = null; } + upgradeHandler = null; sslAttributes = false; asyncContext = null; @@ -3312,14 +3320,34 @@ return sessionId; } + @SuppressWarnings("unchecked") @Override - public <T extends HttpUpgradeHandler> T upgrade(Class<T> arg0) + public <T extends HttpUpgradeHandler> T upgrade(Class<T> upgradeHandlerClass) throws IOException { - // TODO Auto-generated method stub - return null; + T ugradeHandler = null; + Throwable upgradeError = null; + try { + ugradeHandler = (T) context.getInstanceManager().newInstance(upgradeHandlerClass); + } catch (Throwable t) { + upgradeError = t; + } + if (ugradeHandler == null) { + throw new IOException(MESSAGES.upgradeError(), upgradeError); + } + response.sendUpgrade(); + ugradeHandler.init(getEvent()); + this.upgradeHandler = ugradeHandler; + return ugradeHandler; } + public HttpUpgradeHandler getUpgradeHandler() { + return upgradeHandler; + } + public ReadListener getReadListener() { + return inputBuffer.getReadListener(); + } + public String toString() { StringBuilder buf = new StringBuilder(); buf.append("Current Servlet stack for thread ").append(Thread.currentThread().getName()); Modified: branches/8.0.x/src/main/java/org/apache/catalina/connector/Response.java =================================================================== --- branches/8.0.x/src/main/java/org/apache/catalina/connector/Response.java 2013-03-18 16:16:55 UTC (rev 2176) +++ branches/8.0.x/src/main/java/org/apache/catalina/connector/Response.java 2013-03-20 17:30:48 UTC (rev 2177) @@ -40,6 +40,7 @@ import javax.servlet.ServletOutputStream; import javax.servlet.SessionTrackingMode; +import javax.servlet.WriteListener; import javax.servlet.http.Cookie; import javax.servlet.http.HttpServletResponse; @@ -1339,9 +1340,6 @@ if (!connector.hasIoEvents()) throw MESSAGES.cannotUpgradeWithoutEvents(); - if (!request.isEventMode() || request.getAsyncContext() != null) - throw MESSAGES.cannotUpgradeWithoutEventServlet(); - // Ignore any call from an included servlet if (included) return; @@ -1360,9 +1358,6 @@ if (!connector.hasIoEvents()) throw MESSAGES.cannotUpgradeWithoutEvents(); - if (!request.isEventMode() || request.getAsyncContext() != null) - throw MESSAGES.cannotUpgradeWithoutEventServlet(); - // Ignore any call from an included servlet if (included) return; @@ -1797,5 +1792,8 @@ } -} + public WriteListener getWriteListener() { + return outputBuffer.getWriteListener(); + } +} Modified: branches/8.0.x/src/main/java/org/apache/catalina/core/StandardWrapperValve.java =================================================================== --- branches/8.0.x/src/main/java/org/apache/catalina/core/StandardWrapperValve.java 2013-03-18 16:16:55 UTC (rev 2176) +++ branches/8.0.x/src/main/java/org/apache/catalina/core/StandardWrapperValve.java 2013-03-20 17:30:48 UTC (rev 2177) @@ -509,8 +509,8 @@ if (event.getType() == EventType.END || event.getType() == EventType.ERROR || event.getType() == EventType.TIMEOUT) { // Invoke the listeners with onComplete or onTimeout - boolean timeout = (event.getType() == EventType.TIMEOUT) ? true : false; - boolean error = (event.getType() == EventType.ERROR) ? true : false; + boolean timeout = (event.getType() == EventType.TIMEOUT); + boolean error = (event.getType() == EventType.ERROR); Iterator<AsyncListenerRegistration> asyncListenerRegistrations = asyncContext.getAsyncListeners().values().iterator(); while (asyncListenerRegistrations.hasNext()) { @@ -535,12 +535,67 @@ exception(request, response, e); } } + boolean end = (event.getType() == EventType.END) || error; if (timeout && request.isEventMode() && asyncContext.getPath() == null) { response.setStatus(HttpServletResponse.SC_INTERNAL_SERVER_ERROR); + end = true; } + try { + // Call error notifications for IO listeners + if (error) { + if (request.getReadListener() != null) { + request.getReadListener().onError(asyncContext.getError()); + } + if (response.getWriteListener() != null) { + response.getWriteListener().onError(asyncContext.getError()); + } + } + } catch (Throwable e) { + container.getLogger().error(MESSAGES.ioListenerError(getContainer().getName()), e); + exception(request, response, e); + } + if (end && (request.getUpgradeHandler() != null)) { + try { + // FIXME: Examine if need to call elsewhere + request.getUpgradeHandler().destroy(); + } catch (Throwable e) { + container.getLogger().error(MESSAGES.upgradeHandlerDestroyError(getContainer().getName()), e); + exception(request, response, e); + } + } if (error && request.isEventMode() && asyncContext.getPath() == null) { exception(request, response, asyncContext.getError()); } + } else if (event.getType() == EventType.READ) { + // Read notification + if (request.getReadListener() != null) { + try { + request.getReadListener().onDataAvailable(); + } catch (Throwable e) { + container.getLogger().error(MESSAGES.ioListenerError(getContainer().getName()), e); + exception(request, response, e); + } + } + } else if (event.getType() == EventType.EOF) { + // End of stream notification + if (request.getReadListener() != null) { + try { + request.getReadListener().onAllDataRead(); + } catch (Throwable e) { + container.getLogger().error(MESSAGES.ioListenerError(getContainer().getName()), e); + exception(request, response, e); + } + } + } else if (event.getType() == EventType.WRITE) { + // Write notification + if (response.getWriteListener() != null) { + try { + response.getWriteListener().onWritePossible(); + } catch (Throwable e) { + container.getLogger().error(MESSAGES.ioListenerError(getContainer().getName()), e); + exception(request, response, e); + } + } } else if (asyncContext.getRunnable() != null) { // Execute the runnable try { Modified: branches/8.0.x/src/main/java/org/apache/coyote/ActionCode.java =================================================================== --- branches/8.0.x/src/main/java/org/apache/coyote/ActionCode.java 2013-03-18 16:16:55 UTC (rev 2176) +++ branches/8.0.x/src/main/java/org/apache/coyote/ActionCode.java 2013-03-20 17:30:48 UTC (rev 2177) @@ -181,6 +181,16 @@ */ public static final ActionCode ACTION_EVENT_WAKEUP = new ActionCode(29); + /** + * Ask for read callbacks + */ + public static final ActionCode ACTION_EVENT_READ_BEGIN = new ActionCode(30); + + /** + * Ask for a write callbacks + */ + public static final ActionCode ACTION_EVENT_WRITE_BEGIN = new ActionCode(31); + // ----------------------------------------------------------- Constructors int code; Modified: branches/8.0.x/src/main/java/org/apache/coyote/http11/Http11AprProcessor.java =================================================================== --- branches/8.0.x/src/main/java/org/apache/coyote/http11/Http11AprProcessor.java 2013-03-18 16:16:55 UTC (rev 2176) +++ branches/8.0.x/src/main/java/org/apache/coyote/http11/Http11AprProcessor.java 2013-03-20 17:30:48 UTC (rev 2177) @@ -1288,6 +1288,17 @@ writeNotification = true; } else if (actionCode == ActionCode.ACTION_EVENT_TIMEOUT) { timeout = ((Integer) param).intValue(); + } else if (actionCode == ActionCode.ACTION_EVENT_READ_BEGIN) { + Socket.timeoutSet(socket, 0); + inputBuffer.setNonBlocking(true); + readNotifications = true; + } else if (actionCode == ActionCode.ACTION_EVENT_WRITE_BEGIN) { + Socket.timeoutSet(socket, 0); + outputBuffer.setNonBlocking(true); + if (!eventProcessing && !writeNotification) { + endpoint.getEventPoller().add(socket, timeout, false, true, false, true); + } + writeNotification = true; } else if (actionCode == ActionCode.UPGRADE) { // Switch to raw bytes mode inputBuffer.removeActiveFilters(); Modified: branches/8.0.x/src/main/java/org/apache/coyote/http11/Http11NioProcessor.java =================================================================== --- branches/8.0.x/src/main/java/org/apache/coyote/http11/Http11NioProcessor.java 2013-03-18 16:16:55 UTC (rev 2176) +++ branches/8.0.x/src/main/java/org/apache/coyote/http11/Http11NioProcessor.java 2013-03-20 17:30:48 UTC (rev 2177) @@ -38,6 +38,7 @@ import org.apache.coyote.http11.filters.SavedRequestInputFilter; import org.apache.coyote.http11.filters.VoidInputFilter; import org.apache.coyote.http11.filters.VoidOutputFilter; +import org.apache.tomcat.jni.Socket; import org.apache.tomcat.util.buf.ByteChunk; import org.apache.tomcat.util.buf.HexUtils; import org.apache.tomcat.util.buf.MessageBytes; @@ -855,6 +856,12 @@ } else if (actionCode == ActionCode.ACTION_EVENT_TIMEOUT) { // Timeout event timeoutEvent(param); + } else if (actionCode == ActionCode.ACTION_EVENT_READ_BEGIN) { + inputBuffer.setNonBlocking(true); + readNotifications = true; + } else if (actionCode == ActionCode.ACTION_EVENT_WRITE_BEGIN) { + outputBuffer.setNonBlocking(true); + writeEvent(param); } else if (actionCode == ActionCode.UPGRADE) { // Switch to raw bytes mode inputBuffer.removeActiveFilters(); Modified: branches/8.0.x/src/main/java/org/jboss/web/CatalinaMessages.java =================================================================== --- branches/8.0.x/src/main/java/org/jboss/web/CatalinaMessages.java 2013-03-18 16:16:55 UTC (rev 2176) +++ branches/8.0.x/src/main/java/org/jboss/web/CatalinaMessages.java 2013-03-20 17:30:48 UTC (rev 2177) @@ -141,9 +141,6 @@ @Message(id = 34, value = "Cannot upgrade from HTTP/1.1 without IO events") IllegalStateException cannotUpgradeWithoutEvents(); - @Message(id = 35, value = "Cannot upgrade from HTTP/1.1 is not using an HttpEventServlet") - IllegalStateException cannotUpgradeWithoutEventServlet(); - @Message(id = 36, value = "Cannot call sendFile() after the response has been committed") IllegalStateException cannotSendFile(); @@ -975,4 +972,19 @@ @Message(id = 360, value = "No session") IllegalStateException nullSession(); + @Message(id = 361, value = "Read listener already set") + IllegalStateException readListenerAlreadySet(); + + @Message(id = 362, value = "Write listener already set") + IllegalStateException writeListenerAlreadySet(); + + @Message(id = 363, value = "Upgrade error") + String upgradeError(); + + @Message(id = 364, value = "Upgrade destroy processing for servlet %s threw exception") + String upgradeHandlerDestroyError(String servletName); + + @Message(id = 365, value = "IO listener processing for servlet %s threw exception") + String ioListenerError(String servletName); + }