Author: remy.maucherat(a)jboss.com Date: 2013-06-17 10:08:53 -0400 (Mon, 17 Jun 2013) New Revision: 2223 Added: branches/7.2.x/src/main/java/org/apache/catalina/websocket/ branches/7.2.x/src/main/java/org/apache/catalina/websocket/Constants.java branches/7.2.x/src/main/java/org/apache/catalina/websocket/MessageInbound.java branches/7.2.x/src/main/java/org/apache/catalina/websocket/StreamInbound.java branches/7.2.x/src/main/java/org/apache/catalina/websocket/WebSocketServlet.java branches/7.2.x/src/main/java/org/apache/catalina/websocket/WsFrame.java branches/7.2.x/src/main/java/org/apache/catalina/websocket/WsHttpServletRequestWrapper.java branches/7.2.x/src/main/java/org/apache/catalina/websocket/WsInputStream.java branches/7.2.x/src/main/java/org/apache/catalina/websocket/WsOutbound.java branches/7.2.x/src/main/java/org/apache/tomcat/util/buf/Utf8Decoder.java Modified: branches/7.2.x/src/main/java/org/jboss/web/CatalinaMessages.java Log: Port the (basic) websocket API from Tomcat 7 Added: branches/7.2.x/src/main/java/org/apache/catalina/websocket/Constants.java =================================================================== --- branches/7.2.x/src/main/java/org/apache/catalina/websocket/Constants.java (rev 0) +++ branches/7.2.x/src/main/java/org/apache/catalina/websocket/Constants.java 2013-06-17 14:08:53 UTC (rev 2223) @@ -0,0 +1,123 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.catalina.websocket; + +/** + * Constants for this Java package. + */ +public class Constants { + public static final String Package = "org.apache.catalina.websocket"; + + public static final String INBOUND_ATTRIBUTE = "websocket.inbound"; + + // OP Codes + public static final byte OPCODE_CONTINUATION = 0x00; + public static final byte OPCODE_TEXT = 0x01; + public static final byte OPCODE_BINARY = 0x02; + public static final byte OPCODE_CLOSE = 0x08; + public static final byte OPCODE_PING = 0x09; + public static final byte OPCODE_PONG = 0x0A; + + // Status Codes + // Definitions as per RFC 6455 (http://tools.ietf.org/html/rfc6455) + /** + * 1000 indicates a normal closure, meaning whatever purpose the + * connection was established for has been fulfilled. + */ + public static final int STATUS_CLOSE_NORMAL = 1000; + + /** + * 1001 indicates that an endpoint is "going away", such as a server + * going down, or a browser having navigated away from a page. + */ + public static final int STATUS_SHUTDOWN = 1001; + + /** + * 1002 indicates that an endpoint is terminating the connection due + * to a protocol error. + */ + public static final int STATUS_PROTOCOL_ERROR = 1002; + + /** + * 1003 indicates that an endpoint is terminating the connection + * because it has received a type of data it cannot accept (e.g. an + * endpoint that understands only text data MAY send this if it + * receives a binary message). + */ + public static final int STATUS_UNEXPECTED_DATA_TYPE = 1003; + + // 1004 is reserved. The specific meaning might be defined in the future. + + /** + * 1005 is a reserved value and MUST NOT be set as a status code in a + * Close control frame by an endpoint. It is designated for use in + * applications expecting a status code to indicate that no status + * code was actually present. + */ + public static final int STATUS_CODE_MISSING = 1005; + + /** + * 1006 is a reserved value and MUST NOT be set as a status code in a + * Close control frame by an endpoint. It is designated for use in + * applications expecting a status code to indicate that the + * connection was closed abnormally, e.g. without sending or + * receiving a Close control frame. + */ + public static final int STATUS_CLOSED_UNEXPECTEDLY = 1006; + + /** + * 1007 indicates that an endpoint is terminating the connection + * because it has received data within a message that was not + * consistent with the type of the message (e.g., non-UTF-8 [RFC3629] + * data within a text message). + */ + public static final int STATUS_BAD_DATA = 1007; + + /** + * 1008 indicates that an endpoint is terminating the connection + * because it has received a message that violates its policy. This + * is a generic status code that can be returned when there is no + * other more suitable status code (e.g. 1003 or 1009), or if there + * is a need to hide specific details about the policy. + */ + public static final int STATUS_POLICY_VIOLATION = 1008; + + /** + * 1009 indicates that an endpoint is terminating the connection + * because it has received a message which is too big for it to + * process. + */ + public static final int STATUS_MESSAGE_TOO_LARGE = 1009; + + /** + * 1010 indicates that an endpoint (client) is terminating the + * connection because it has expected the server to negotiate one or + * more extension, but the server didn't return them in the response + * message of the WebSocket handshake. The list of extensions which + * are needed SHOULD appear in the /reason/ part of the Close frame. + * Note that this status code is not used by the server, because it + * can fail the WebSocket handshake instead. + */ + public static final int STATUS_REQUIRED_EXTENSION = 1010; + + /** + * 1011 indicates that a server is terminating the connection because it + * encountered an unexpected condition that prevented it from fulfilling the + * request. + */ + public static final int STATUS_UNEXPECTED_CONDITION = 1011; +} Added: branches/7.2.x/src/main/java/org/apache/catalina/websocket/MessageInbound.java =================================================================== --- branches/7.2.x/src/main/java/org/apache/catalina/websocket/MessageInbound.java (rev 0) +++ branches/7.2.x/src/main/java/org/apache/catalina/websocket/MessageInbound.java 2013-06-17 14:08:53 UTC (rev 2223) @@ -0,0 +1,178 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.catalina.websocket; + +import static org.jboss.web.CatalinaMessages.MESSAGES; + +import java.io.IOException; +import java.io.InputStream; +import java.io.Reader; +import java.nio.ByteBuffer; +import java.nio.CharBuffer; + +/** + * Base implementation of the class used to process WebSocket connections based + * on messages. Applications should extend this class to provide application + * specific functionality. Applications that wish to operate on a stream basis + * rather than a message basis should use {@link StreamInbound}. + */ +public abstract class MessageInbound extends StreamInbound { + + // 2MB - like maxPostSize + private int byteBufferMaxSize = 2097152; + private int charBufferMaxSize = 2097152; + + private ByteBuffer bb = ByteBuffer.allocate(8192); + private CharBuffer cb = CharBuffer.allocate(8192); + + + @Override + protected final void onBinaryData(InputStream is) throws IOException { + int read = 0; + while (read > -1) { + bb.position(bb.position() + read); + if (bb.remaining() == 0) { + resizeByteBuffer(); + } + read = is.read(bb.array(), bb.position(), bb.remaining()); + } + bb.flip(); + onBinaryMessage(bb); + bb.clear(); + } + + + @Override + protected final void onTextData(Reader r) throws IOException { + int read = 0; + while (read > -1) { + cb.position(cb.position() + read); + if (cb.remaining() == 0) { + resizeCharBuffer(); + } + read = r.read(cb.array(), cb.position(), cb.remaining()); + } + cb.flip(); + onTextMessage(cb); + cb.clear(); + } + + + private void resizeByteBuffer() throws IOException { + int maxSize = getByteBufferMaxSize(); + if (bb.limit() >= maxSize) { + throw new IOException(MESSAGES.bufferTooSmall()); + } + + long newSize = bb.limit() * 2; + if (newSize > maxSize) { + newSize = maxSize; + } + + // Cast is safe. newSize < maxSize and maxSize is an int + ByteBuffer newBuffer = ByteBuffer.allocate((int) newSize); + bb.rewind(); + newBuffer.put(bb); + bb = newBuffer; + } + + + private void resizeCharBuffer() throws IOException { + int maxSize = getCharBufferMaxSize(); + if (cb.limit() >= maxSize) { + throw new IOException(MESSAGES.bufferTooSmall()); + } + + long newSize = cb.limit() * 2; + if (newSize > maxSize) { + newSize = maxSize; + } + + // Cast is safe. newSize < maxSize and maxSize is an int + CharBuffer newBuffer = CharBuffer.allocate((int) newSize); + cb.rewind(); + newBuffer.put(cb); + cb = newBuffer; + } + + + /** + * Obtain the current maximum size (in bytes) of the buffer used for binary + * messages. + */ + public final int getByteBufferMaxSize() { + return byteBufferMaxSize; + } + + + /** + * Set the maximum size (in bytes) of the buffer used for binary messages. + */ + public final void setByteBufferMaxSize(int byteBufferMaxSize) { + this.byteBufferMaxSize = byteBufferMaxSize; + } + + + /** + * Obtain the current maximum size (in characters) of the buffer used for + * binary messages. + */ + public final int getCharBufferMaxSize() { + return charBufferMaxSize; + } + + + /** + * Set the maximum size (in characters) of the buffer used for textual + * messages. + */ + public final void setCharBufferMaxSize(int charBufferMaxSize) { + this.charBufferMaxSize = charBufferMaxSize; + } + + + /** + * This method is called when there is a binary WebSocket message available + * to process. The message is presented via a ByteBuffer and may have been + * formed from one or more frames. The number of frames used to transmit the + * message is not made visible to the application. + * + * @param message The WebSocket message + * + * @throws IOException If a problem occurs processing the message. Any + * exception will trigger the closing of the WebSocket + * connection. + */ + protected abstract void onBinaryMessage(ByteBuffer message) + throws IOException; + + + /** + * This method is called when there is a textual WebSocket message available + * to process. The message is presented via a CharBuffer and may have been + * formed from one or more frames. The number of frames used to transmit the + * message is not made visible to the application. + * + * @param message The WebSocket message + * + * @throws IOException If a problem occurs processing the message. Any + * exception will trigger the closing of the WebSocket + * connection. + */ + protected abstract void onTextMessage(CharBuffer message) + throws IOException; +} Added: branches/7.2.x/src/main/java/org/apache/catalina/websocket/StreamInbound.java =================================================================== --- branches/7.2.x/src/main/java/org/apache/catalina/websocket/StreamInbound.java (rev 0) +++ branches/7.2.x/src/main/java/org/apache/catalina/websocket/StreamInbound.java 2013-06-17 14:08:53 UTC (rev 2223) @@ -0,0 +1,314 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.catalina.websocket; + +import java.io.IOException; +import java.io.InputStream; +import java.io.InputStreamReader; +import java.io.Reader; +import java.nio.ByteBuffer; +import java.nio.charset.MalformedInputException; +import java.nio.charset.UnmappableCharacterException; + +import org.apache.tomcat.util.buf.Utf8Decoder; +import org.jboss.servlet.http.HttpEvent; + +/** + * Base implementation of the class used to process WebSocket connections based + * on streams. Applications should extend this class to provide application + * specific functionality. Applications that wish to operate on a message basis + * rather than a stream basis should use {@link MessageInbound}. + */ +public abstract class StreamInbound { + + private final ClassLoader applicationClassLoader; + private HttpEvent event = null; + private WsOutbound outbound; + private int outboundByteBufferSize = WsOutbound.DEFAULT_BUFFER_SIZE; + private int outboundCharBufferSize = WsOutbound.DEFAULT_BUFFER_SIZE; + + + public StreamInbound() { + applicationClassLoader = Thread.currentThread().getContextClassLoader(); + } + + + public int getOutboundByteBufferSize() { + return outboundByteBufferSize; + } + + public HttpEvent getEvent() { + return event; + } + + public void setEvent(HttpEvent event) { + this.event = event; + } + + /** + * This only applies to the {@link WsOutbound} instance returned from + * {@link #getWsOutbound()} created by a subsequent call to + * {@link #setUpgradeOutbound(UpgradeOutbound)}. The current + * {@link WsOutbound} instance, if any, is not affected. + * + * @param outboundByteBufferSize + */ + public void setOutboundByteBufferSize(int outboundByteBufferSize) { + this.outboundByteBufferSize = outboundByteBufferSize; + } + + + public int getOutboundCharBufferSize() { + return outboundCharBufferSize; + } + + + /** + * This only applies to the {@link WsOutbound} instance returned from + * {@link #getWsOutbound()} created by a subsequent call to + * {@link #setUpgradeOutbound(UpgradeOutbound)}. The current + * {@link WsOutbound} instance, if any, is not affected. + * + * @param outboundCharBufferSize + */ + public void setOutboundCharBufferSize(int outboundCharBufferSize) { + this.outboundCharBufferSize = outboundCharBufferSize; + } + + + /** + * Obtain the outbound side of this WebSocket connection used for writing + * data to the client. + */ + public final WsOutbound getWsOutbound() { + return outbound; + } + + public final void onData() throws IOException { + // Must be start the start of a message (which may consist of multiple + // frames) + WsInputStream wsIs = new WsInputStream(event, getWsOutbound()); + + try { + WsFrame frame = wsIs.nextFrame(false); + + while (frame != null) { + // TODO User defined extensions may define values for rsv + if (frame.getRsv() > 0) { + closeOutboundConnection( + Constants.STATUS_PROTOCOL_ERROR, null); + event.close(); + } + + byte opCode = frame.getOpCode(); + + if (opCode == Constants.OPCODE_BINARY) { + doOnBinaryData(wsIs); + } else if (opCode == Constants.OPCODE_TEXT) { + InputStreamReader r = + new InputStreamReader(wsIs, new Utf8Decoder()); + doOnTextData(r); + } else if (opCode == Constants.OPCODE_CLOSE){ + closeOutboundConnection(frame); + event.close(); + } else if (opCode == Constants.OPCODE_PING) { + getWsOutbound().pong(frame.getPayLoad()); + } else if (opCode == Constants.OPCODE_PONG) { + doOnPong(frame.getPayLoad()); + } else { + // Unknown OpCode + closeOutboundConnection( + Constants.STATUS_PROTOCOL_ERROR, null); + event.close(); + } + frame = wsIs.nextFrame(false); + } + } catch (MalformedInputException mie) { + // Invalid UTF-8 + closeOutboundConnection(Constants.STATUS_BAD_DATA, null); + event.close(); + } catch (UnmappableCharacterException uce) { + // Invalid UTF-8 + closeOutboundConnection(Constants.STATUS_BAD_DATA, null); + event.close(); + } catch (IOException ioe) { + // Given something must have gone to reach this point, this + // might not work but try it anyway. + closeOutboundConnection(Constants.STATUS_PROTOCOL_ERROR, null); + event.close(); + } + } + + private void doOnBinaryData(InputStream is) throws IOException { + // Need to call onBinaryData using the web application's class loader + Thread t = Thread.currentThread(); + ClassLoader cl = t.getContextClassLoader(); + t.setContextClassLoader(applicationClassLoader); + try { + onBinaryData(is); + } finally { + t.setContextClassLoader(cl); + } + } + + + private void doOnTextData(Reader r) throws IOException { + // Need to call onTextData using the web application's class loader + Thread t = Thread.currentThread(); + ClassLoader cl = t.getContextClassLoader(); + t.setContextClassLoader(applicationClassLoader); + try { + onTextData(r); + } finally { + t.setContextClassLoader(cl); + } + } + + + private void closeOutboundConnection(int status, ByteBuffer data) throws IOException { + try { + getWsOutbound().close(status, data); + } finally { + doOnClose(status); + } + } + + private void closeOutboundConnection(WsFrame frame) throws IOException { + try { + getWsOutbound().close(frame); + } finally { + doOnClose(Constants.STATUS_CLOSE_NORMAL); + } + } + + /** + * Package private so the outbound connection can signal that the connection + * has been closed - usually due to an error. + * + * @param status The WebSocket status code to report to the application + */ + void doOnClose(int status) { + // Need to call onClose using the web application's class loader + Thread t = Thread.currentThread(); + ClassLoader cl = t.getContextClassLoader(); + t.setContextClassLoader(applicationClassLoader); + try { + onClose(status); + } finally { + t.setContextClassLoader(cl); + } + } + + private void doOnPong(ByteBuffer payload) { + // Need to call onPong using the web application's class loader + Thread t = Thread.currentThread(); + ClassLoader cl = t.getContextClassLoader(); + t.setContextClassLoader(applicationClassLoader); + try { + onPong(payload); + } finally { + t.setContextClassLoader(cl); + } + } + + public final void onUpgradeComplete() { + outbound = new WsOutbound(event, this, outboundByteBufferSize, + outboundCharBufferSize); + // Need to call onOpen using the web application's class loader + Thread t = Thread.currentThread(); + ClassLoader cl = t.getContextClassLoader(); + t.setContextClassLoader(applicationClassLoader); + try { + onOpen(outbound); + } finally { + t.setContextClassLoader(cl); + } + } + + /** + * Intended to be overridden by sub-classes that wish to be notified + * when the outbound connection is established. The default implementation + * is a NO-OP. + * + * @param outbound The outbound WebSocket connection. + */ + protected void onOpen(WsOutbound outbound) { + // NO-OP + } + + /** + * Intended to be overridden by sub-classes that wish to be notified + * when the outbound connection is closed. The default implementation + * is a NO-OP. + * + * @param status The status code of the close reason. + */ + protected void onClose(int status) { + // NO-OP + } + + /** + * Intended to be overridden by sub-classes that wish to be notified + * when a pong is received. The default implementation is a NO-OP. + * + * @param payload The payload included in the pong. + */ + protected void onPong(ByteBuffer payload) { + // NO-OP + } + + /** + * This method is called when there is a binary WebSocket message available + * to process. The message is presented via a stream and may be formed from + * one or more frames. The number of frames used to transmit the message is + * not made visible to the application. + * + * @param is The WebSocket message + * + * @throws IOException If a problem occurs processing the message. Any + * exception will trigger the closing of the WebSocket + * connection. + */ + protected abstract void onBinaryData(InputStream is) throws IOException; + + + /** + * This method is called when there is a textual WebSocket message available + * to process. The message is presented via a reader and may be formed from + * one or more frames. The number of frames used to transmit the message is + * not made visible to the application. + * + * @param r The WebSocket message + * + * @throws IOException If a problem occurs processing the message. Any + * exception will trigger the closing of the WebSocket + * connection. + */ + protected abstract void onTextData(Reader r) throws IOException; + + /** + * This default implementation sets the read timeout to infinite and expects + * the WebSocket application to close the connection when it is no longer + * required. Applications wishing to set an explicit timeout may override + * this method and return a value of their choice. + * + * @return The read timeout in milliseconds or -1 for infinite + */ + public int getReadTimeout() { + return -1; + } +} Added: branches/7.2.x/src/main/java/org/apache/catalina/websocket/WebSocketServlet.java =================================================================== --- branches/7.2.x/src/main/java/org/apache/catalina/websocket/WebSocketServlet.java (rev 0) +++ branches/7.2.x/src/main/java/org/apache/catalina/websocket/WebSocketServlet.java 2013-06-17 14:08:53 UTC (rev 2223) @@ -0,0 +1,281 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.catalina.websocket; + +import static org.jboss.web.CatalinaMessages.MESSAGES; + +import java.io.IOException; +import java.security.MessageDigest; +import java.security.NoSuchAlgorithmException; +import java.util.ArrayList; +import java.util.Collections; +import java.util.Enumeration; +import java.util.List; +import java.util.Queue; +import java.util.concurrent.ConcurrentLinkedQueue; + +import javax.servlet.ServletException; +import javax.servlet.ServletResponse; +import javax.servlet.ServletResponseWrapper; +import javax.servlet.http.HttpServlet; +import javax.servlet.http.HttpServletRequest; +import javax.servlet.http.HttpServletResponse; + +import org.apache.catalina.util.Base64; +import org.apache.tomcat.util.buf.EncodingToCharset; +import org.jboss.servlet.http.HttpEvent; +import org.jboss.servlet.http.HttpEventServlet; +import org.jboss.servlet.http.UpgradableHttpServletResponse; + +/** + * Provides the base implementation of a Servlet for processing WebSocket + * connections as per RFC6455. It is expected that applications will extend this + * implementation and provide application specific functionality. + */ +public abstract class WebSocketServlet extends HttpServlet implements HttpEventServlet { + + private static final long serialVersionUID = 1L; + private static final byte[] WS_ACCEPT = + "258EAFA5-E914-47DA-95CA-C5AB0DC85B11".getBytes( + EncodingToCharset.ISO_8859_1); + + private final Queue<MessageDigest> sha1Helpers = + new ConcurrentLinkedQueue<MessageDigest>(); + + + public void event(HttpEvent event) throws IOException, ServletException { + + HttpServletRequest req = event.getHttpServletRequest(); + HttpServletResponse resp = event.getHttpServletResponse(); + switch (event.getType()) { + case BEGIN: + // Information required to send the server handshake message + String key; + String subProtocol = null; + List<String> extensions = Collections.emptyList(); + + if (!headerContainsToken(req, "upgrade", "websocket")) { + resp.sendError(HttpServletResponse.SC_BAD_REQUEST); + return; + } + + if (!headerContainsToken(req, "connection", "upgrade")) { + resp.sendError(HttpServletResponse.SC_BAD_REQUEST); + return; + } + + if (!headerContainsToken(req, "sec-websocket-version", "13")) { + resp.setStatus(426); + resp.setHeader("Sec-WebSocket-Version", "13"); + return; + } + + key = req.getHeader("Sec-WebSocket-Key"); + if (key == null) { + resp.sendError(HttpServletResponse.SC_BAD_REQUEST); + return; + } + + String origin = req.getHeader("Origin"); + if (!verifyOrigin(origin)) { + resp.sendError(HttpServletResponse.SC_FORBIDDEN); + return; + } + + List<String> subProtocols = getTokensFromHeader(req, + "Sec-WebSocket-Protocol"); + if (!subProtocols.isEmpty()) { + subProtocol = selectSubProtocol(subProtocols); + + } + + // Small hack until the Servlet API provides a way to do this. + ServletResponse inner = resp; + // Unwrap the request + while (inner instanceof ServletResponseWrapper) { + inner = ((ServletResponseWrapper) inner).getResponse(); + } + if (inner instanceof UpgradableHttpServletResponse) { + ((UpgradableHttpServletResponse) inner).startUpgrade(); + } else { + resp.sendError(HttpServletResponse.SC_INTERNAL_SERVER_ERROR, + MESSAGES.errorUpgrading()); + } + + // TODO Read client handshake - Sec-WebSocket-Extensions + + // TODO Extensions require the ability to specify something (API TBD) + // that can be passed to the Tomcat internals and process extension + // data present when the frame is fragmented. + + // If we got this far, all is good. Accept the connection. + resp.setHeader("Upgrade", "websocket"); + resp.setHeader("Connection", "upgrade"); + resp.setHeader("Sec-WebSocket-Accept", getWebSocketAccept(key)); + if (subProtocol != null) { + resp.setHeader("Sec-WebSocket-Protocol", subProtocol); + } + if (!extensions.isEmpty()) { + // TODO + } + + WsHttpServletRequestWrapper wrapper = new WsHttpServletRequestWrapper(req); + StreamInbound inbound = createWebSocketInbound(subProtocol, wrapper); + inbound.setEvent(event); + wrapper.invalidate(); + + ((UpgradableHttpServletResponse) inner).sendUpgrade(); + + inbound.onUpgradeComplete(); + req.setAttribute(Constants.INBOUND_ATTRIBUTE, inbound); + break; + case END: + event.close(); + break; + case EOF: + break; + case ERROR: + event.close(); + break; + case EVENT: + break; + case READ: + inbound = (StreamInbound) req.getAttribute(Constants.INBOUND_ATTRIBUTE); + do { + inbound.onData(); + } while (event.isReadReady()); + break; + case TIMEOUT: + // Ignore ? + break; + case WRITE: + break; + } + + } + + + /* + * This only works for tokens. Quoted strings need more sophisticated + * parsing. + */ + private boolean headerContainsToken(HttpServletRequest req, + String headerName, String target) { + Enumeration<String> headers = req.getHeaders(headerName); + while (headers.hasMoreElements()) { + String header = headers.nextElement(); + String[] tokens = header.split(","); + for (String token : tokens) { + if (target.equalsIgnoreCase(token.trim())) { + return true; + } + } + } + return false; + } + + + /* + * This only works for tokens. Quoted strings need more sophisticated + * parsing. + */ + private List<String> getTokensFromHeader(HttpServletRequest req, + String headerName) { + List<String> result = new ArrayList<String>(); + + Enumeration<String> headers = req.getHeaders(headerName); + while (headers.hasMoreElements()) { + String header = headers.nextElement(); + String[] tokens = header.split(","); + for (String token : tokens) { + result.add(token.trim()); + } + } + return result; + } + + + private String getWebSocketAccept(String key) throws ServletException { + + MessageDigest sha1Helper = sha1Helpers.poll(); + if (sha1Helper == null) { + try { + sha1Helper = MessageDigest.getInstance("SHA1"); + } catch (NoSuchAlgorithmException e) { + throw new ServletException(e); + } + } + + sha1Helper.reset(); + sha1Helper.update(key.getBytes(EncodingToCharset.ISO_8859_1)); + String result = new String(Base64.encode(sha1Helper.digest(WS_ACCEPT))); + + sha1Helpers.add(sha1Helper); + + return result; + } + + + /** + * Intended to be overridden by sub-classes that wish to verify the origin + * of a WebSocket request before processing it. + * + * @param origin The value of the origin header from the request which + * may be <code>null</code> + * + * @return <code>true</code> to accept the request. <code>false</code> to + * reject it. This default implementation always returns + * <code>true</code>. + */ + protected boolean verifyOrigin(String origin) { + return true; + } + + + /** + * Intended to be overridden by sub-classes that wish to select a + * sub-protocol if the client provides a list of supported protocols. + * + * @param subProtocols The list of sub-protocols supported by the client + * in client preference order. The server is under no + * obligation to respect the declared preference + * @return <code>null</code> if no sub-protocol is selected or the name of + * the protocol which <b>must</b> be one of the protocols listed by + * the client. This default implementation always returns + * <code>null</code>. + */ + protected String selectSubProtocol(List<String> subProtocols) { + return null; + } + + + /** + * Create the instance that will process this inbound connection. + * Applications must provide a new instance for each connection. + * + * @param subProtocol The sub-protocol agreed between the client and + * server or <code>null</code> if none was agreed + * @param request The HTTP request that initiated this WebSocket + * connection. Note that this object is <b>only</b> + * valid inside this method. You must not retain a + * reference to it outside the execution of this + * method. If Tomcat detects such access, it will throw + * an IllegalStateException + */ + protected abstract StreamInbound createWebSocketInbound(String subProtocol, + HttpServletRequest request); +} Added: branches/7.2.x/src/main/java/org/apache/catalina/websocket/WsFrame.java =================================================================== --- branches/7.2.x/src/main/java/org/apache/catalina/websocket/WsFrame.java (rev 0) +++ branches/7.2.x/src/main/java/org/apache/catalina/websocket/WsFrame.java 2013-06-17 14:08:53 UTC (rev 2223) @@ -0,0 +1,243 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.catalina.websocket; + +import static org.jboss.web.CatalinaMessages.MESSAGES; + +import java.io.EOFException; +import java.io.IOException; +import java.nio.ByteBuffer; +import java.nio.CharBuffer; +import java.nio.charset.CoderResult; + +import org.apache.tomcat.util.buf.Utf8Decoder; +import org.jboss.servlet.http.HttpEvent; + +/** + * Represents a complete WebSocket frame with the exception of the payload for + * non-control frames. + */ +public class WsFrame { + + private final boolean fin; + private final int rsv; + private final byte opCode; + private final byte[] mask = new byte[4]; + private long payloadLength; + private final ByteBuffer payload; + + /** + * Create the new WebSocket frame, reading data from the processor as + * necessary. + * + * @param first First byte of data for this frame + * @param processor Processor associated with the WebSocket connection on + * which the frame has been sent + * + * @throws IOException If a problem occurs processing the frame. Any + * exception will trigger the closing of the WebSocket + * connection. + */ + private WsFrame(byte first, HttpEvent event) throws IOException { + + int b = first & 0xFF; + fin = (b & 0x80) > 0; + rsv = (b & 0x70) >>> 4; + opCode = (byte) (b & 0x0F); + + b = blockingRead(event); + // Client data must be masked + if ((b & 0x80) == 0) { + throw new IOException(MESSAGES.frameNotMasked()); + } + + payloadLength = b & 0x7F; + if (payloadLength == 126) { + byte[] extended = new byte[2]; + blockingRead(event, extended); + payloadLength = byteArrayToLong(extended); + } else if (payloadLength == 127) { + byte[] extended = new byte[8]; + blockingRead(event, extended); + payloadLength = byteArrayToLong(extended); + } + + if (isControl()) { + if (payloadLength > 125) { + throw new IOException(); + } + if (!fin) { + throw new IOException(); + } + } + + blockingRead(event, mask); + + if (isControl()) { + // Note: Payload limited to <= 125 bytes by test above + payload = ByteBuffer.allocate((int) payloadLength); + blockingRead(event, payload); + + if (opCode == Constants.OPCODE_CLOSE && payloadLength > 2) { + // Check close payload - if present - is valid UTF-8 + CharBuffer cb = CharBuffer.allocate((int) payloadLength); + Utf8Decoder decoder = new Utf8Decoder(); + payload.position(2); + CoderResult cr = decoder.decode(payload, cb, true); + payload.position(0); + if (cr.isError()) { + throw new IOException(MESSAGES.frameInvalidUtf8()); + } + } + } else { + payload = null; + } + } + + public static long byteArrayToLong(byte[] input) throws IOException { + if (input.length > 8) { + // TODO: Better message + throw new IOException(); + } + + int shift = 0; + long result = 0; + for (int i = input.length - 1; i >= 0; i--) { + result = result + ((input[i] & 0xFF) << shift); + shift += 8; + } + + return result; + } + + public boolean getFin() { + return fin; + } + + public int getRsv() { + return rsv; + } + + public byte getOpCode() { + return opCode; + } + + public boolean isControl() { + return (opCode & 0x08) > 0; + } + + public byte[] getMask() { + return mask; + } + + public long getPayLoadLength() { + return payloadLength; + } + + public ByteBuffer getPayLoad() { + return payload; + } + + + /* + * Blocks until a aingle byte has been read + */ + private int blockingRead(HttpEvent event) + throws IOException { + int result = event.getHttpServletRequest().getInputStream().read(); + if (result == -1) { + throw new IOException(MESSAGES.frameEos()); + } + return result; + } + + + /* + * Blocks until the byte array has been filled. + */ + private void blockingRead(HttpEvent event, byte[] bytes) + throws IOException { + int read = 0; + int last = 0; + while (read < bytes.length) { + last = event.getHttpServletRequest().getInputStream().read(bytes, read, bytes.length - read); + if (last == -1) { + throw new IOException(MESSAGES.frameEos()); + } + read += last; + } + } + + + /* + * Intended to read whole payload and blocks until it has. Therefore able to + * unmask the payload data. + */ + private void blockingRead(HttpEvent event, ByteBuffer bb) + throws IOException { + int last = 0; + while (bb.hasRemaining()) { + last = event.getHttpServletRequest().getInputStream().read(); + if (last == -1) { + throw new IOException(MESSAGES.frameEos()); + } + bb.put((byte) (last ^ mask[bb.position() % 4])); + } + bb.flip(); + } + + + /** + * Read the next WebSocket frame, reading data from the processor as + * necessary. + * + * @param processor Processor associated with the WebSocket connection on + * which the frame has been sent + * + * @param block Should this method block until a frame is presented if no + * data is currently available to process. Note that is a + * single byte is available, this method will block until the + * complete frame (excluding payload for non-control frames) is + * available. + * + * @throws IOException If a problem occurs processing the frame. Any + * exception will trigger the closing of the WebSocket + * connection. + */ + public static WsFrame nextFrame(HttpEvent event, + boolean block) throws IOException { + + byte[] first = new byte[1]; + int read = 0; + if (block) { + read = event.getHttpServletRequest().getInputStream().read(first, 0, 1); + } else { + if (event.isReadReady()) { + read = event.getHttpServletRequest().getInputStream().read(first, 0, 1); + } + } + if (read == 1) { + return new WsFrame(first[0], event); + } else if (read == 0) { + return null; + } else if (read == -1) { + throw new EOFException(MESSAGES.frameEos()); + } else { + throw new IOException(MESSAGES.frameFailedRead(Integer.valueOf(read))); + } + } +} Added: branches/7.2.x/src/main/java/org/apache/catalina/websocket/WsHttpServletRequestWrapper.java =================================================================== --- branches/7.2.x/src/main/java/org/apache/catalina/websocket/WsHttpServletRequestWrapper.java (rev 0) +++ branches/7.2.x/src/main/java/org/apache/catalina/websocket/WsHttpServletRequestWrapper.java 2013-06-17 14:08:53 UTC (rev 2223) @@ -0,0 +1,401 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.catalina.websocket; + +import static org.jboss.web.CatalinaMessages.MESSAGES; + +import java.io.BufferedReader; +import java.io.IOException; +import java.io.UnsupportedEncodingException; +import java.security.Principal; +import java.util.Collection; +import java.util.Enumeration; +import java.util.Locale; +import java.util.Map; + +import javax.servlet.AsyncContext; +import javax.servlet.DispatcherType; +import javax.servlet.RequestDispatcher; +import javax.servlet.ServletContext; +import javax.servlet.ServletException; +import javax.servlet.ServletInputStream; +import javax.servlet.ServletRequest; +import javax.servlet.ServletResponse; +import javax.servlet.http.Cookie; +import javax.servlet.http.HttpServletRequest; +import javax.servlet.http.HttpServletResponse; +import javax.servlet.http.HttpSession; +import javax.servlet.http.Part; + +/** + * Wrapper for the HttpServletRequest object that allows the underlying request + * object to be invalidated. + */ +public class WsHttpServletRequestWrapper implements HttpServletRequest { + + private HttpServletRequest request; + + public WsHttpServletRequestWrapper(HttpServletRequest request) { + this.request = request; + } + + private HttpServletRequest getRequest() { + if (request == null) { + throw MESSAGES.invalidWrapper(); + } + return request; + } + + protected void invalidate() { + request = null; + } + + @Override + public Object getAttribute(String name) { + return getRequest().getAttribute(name); + } + + @Override + public Enumeration<String> getAttributeNames() { + return getRequest().getAttributeNames(); + } + + @Override + public String getCharacterEncoding() { + return getRequest().getCharacterEncoding(); + } + + @Override + public void setCharacterEncoding(String env) + throws UnsupportedEncodingException { + getRequest().setCharacterEncoding(env); + } + + @Override + public int getContentLength() { + return getRequest().getContentLength(); + } + + @Override + public String getContentType() { + return getRequest().getContentType(); + } + + @Override + public ServletInputStream getInputStream() throws IOException { + return getRequest().getInputStream(); + } + + @Override + public String getParameter(String name) { + return getRequest().getParameter(name); + } + + @Override + public Enumeration<String> getParameterNames() { + return getRequest().getParameterNames(); + } + + @Override + public String[] getParameterValues(String name) { + return getRequest().getParameterValues(name); + } + + @Override + public Map<String, String[]> getParameterMap() { + return getRequest().getParameterMap(); + } + + @Override + public String getProtocol() { + return getRequest().getProtocol(); + } + + @Override + public String getScheme() { + return getRequest().getScheme(); + } + + @Override + public String getServerName() { + return getRequest().getServerName(); + } + + @Override + public int getServerPort() { + return getRequest().getServerPort(); + } + + @Override + public BufferedReader getReader() throws IOException { + return getRequest().getReader(); + } + + @Override + public String getRemoteAddr() { + return getRequest().getRemoteAddr(); + } + + @Override + public String getRemoteHost() { + return getRequest().getRemoteHost(); + } + + @Override + public void setAttribute(String name, Object o) { + getRequest().setAttribute(name, o); + } + + @Override + public void removeAttribute(String name) { + getRequest().removeAttribute(name); + } + + @Override + public Locale getLocale() { + return getRequest().getLocale(); + } + + @Override + public Enumeration<Locale> getLocales() { + return getRequest().getLocales(); + } + + @Override + public boolean isSecure() { + return getRequest().isSecure(); + } + + @Override + public RequestDispatcher getRequestDispatcher(String path) { + return getRequest().getRequestDispatcher(path); + } + + @Override + @Deprecated + public String getRealPath(String path) { + return getRequest().getRealPath(path); + } + + @Override + public int getRemotePort() { + return getRequest().getRemotePort(); + } + + @Override + public String getLocalName() { + return getRequest().getLocalName(); + } + + @Override + public String getLocalAddr() { + return getRequest().getLocalAddr(); + } + + @Override + public int getLocalPort() { + return getRequest().getLocalPort(); + } + + @Override + public ServletContext getServletContext() { + return getRequest().getServletContext(); + } + + @Override + public AsyncContext startAsync() throws IllegalStateException { + return getRequest().startAsync(); + } + + @Override + public AsyncContext startAsync(ServletRequest servletRequest, + ServletResponse servletResponse) throws IllegalStateException { + return getRequest().startAsync(servletRequest, servletResponse); + } + + @Override + public boolean isAsyncStarted() { + return getRequest().isAsyncStarted(); + } + + @Override + public boolean isAsyncSupported() { + return getRequest().isAsyncSupported(); + } + + @Override + public AsyncContext getAsyncContext() { + return getRequest().getAsyncContext(); + } + + @Override + public DispatcherType getDispatcherType() { + return getRequest().getDispatcherType(); + } + + @Override + public String getAuthType() { + return getRequest().getAuthType(); + } + + @Override + public Cookie[] getCookies() { + return getRequest().getCookies(); + } + + @Override + public long getDateHeader(String name) { + return getRequest().getDateHeader(name); + } + + @Override + public String getHeader(String name) { + return getRequest().getHeader(name); + } + + @Override + public Enumeration<String> getHeaders(String name) { + return getRequest().getHeaders(name); + } + + @Override + public Enumeration<String> getHeaderNames() { + return getRequest().getHeaderNames(); + } + + @Override + public int getIntHeader(String name) { + return getRequest().getIntHeader(name); + } + + @Override + public String getMethod() { + return getRequest().getMethod(); + } + + @Override + public String getPathInfo() { + return getRequest().getPathInfo(); + } + + @Override + public String getPathTranslated() { + return getRequest().getPathTranslated(); + } + + @Override + public String getContextPath() { + return getRequest().getContextPath(); + } + + @Override + public String getQueryString() { + return getRequest().getQueryString(); + } + + @Override + public String getRemoteUser() { + return getRequest().getRemoteUser(); + } + + @Override + public boolean isUserInRole(String role) { + return getRequest().isUserInRole(role); + } + + @Override + public Principal getUserPrincipal() { + return getRequest().getUserPrincipal(); + } + + @Override + public String getRequestedSessionId() { + return getRequest().getRequestedSessionId(); + } + + @Override + public String getRequestURI() { + return getRequest().getRequestURI(); + } + + @Override + public StringBuffer getRequestURL() { + return getRequest().getRequestURL(); + } + + @Override + public String getServletPath() { + return getRequest().getServletPath(); + } + + @Override + public HttpSession getSession(boolean create) { + return getRequest().getSession(create); + } + + @Override + public HttpSession getSession() { + return getRequest().getSession(); + } + + @Override + public boolean isRequestedSessionIdValid() { + return getRequest().isRequestedSessionIdValid(); + } + + @Override + public boolean isRequestedSessionIdFromCookie() { + return getRequest().isRequestedSessionIdFromCookie(); + } + + @Override + public boolean isRequestedSessionIdFromURL() { + return getRequest().isRequestedSessionIdFromURL(); + } + + @Override + @Deprecated + public boolean isRequestedSessionIdFromUrl() { + return getRequest().isRequestedSessionIdFromUrl(); + } + + @Override + public boolean authenticate(HttpServletResponse response) + throws IOException, ServletException { + return getRequest().authenticate(response); + } + + @Override + public void login(String username, String password) throws ServletException { + getRequest().login(username, password); + } + + @Override + public void logout() throws ServletException { + getRequest().logout(); + } + + @Override + public Collection<Part> getParts() throws IOException, ServletException { + return getRequest().getParts(); + } + + @Override + public Part getPart(String name) throws IOException, ServletException { + return getRequest().getPart(name); + } +} Added: branches/7.2.x/src/main/java/org/apache/catalina/websocket/WsInputStream.java =================================================================== --- branches/7.2.x/src/main/java/org/apache/catalina/websocket/WsInputStream.java (rev 0) +++ branches/7.2.x/src/main/java/org/apache/catalina/websocket/WsInputStream.java 2013-06-17 14:08:53 UTC (rev 2223) @@ -0,0 +1,152 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.catalina.websocket; + +import static org.jboss.web.CatalinaMessages.MESSAGES; + +import java.io.IOException; +import java.io.InputStream; + +import org.jboss.servlet.http.HttpEvent; + +/** + * This class is used to read WebSocket frames from the underlying socket and + * makes the payload available for reading as an {@link InputStream}. It only + * makes the number of bytes declared in the payload length available for + * reading even if more bytes are available from the socket. + */ +public class WsInputStream extends InputStream { + + private final HttpEvent event; + private final WsOutbound outbound; + + private WsFrame frame; + private long remaining; + private long readThisFragment; + + private String error = null; + + + public WsInputStream(HttpEvent event, WsOutbound outbound) { + this.event = event; + this.outbound = outbound; + } + + + /** + * Process the next WebSocket frame. + * + * @param block Should this method block until a frame is presented if no + * data is currently available to process. Note that if a + * single byte is available, this method will block until the + * complete frame (excluding payload for non-control frames) is + * available. + * + * @return The next frame to be processed or <code>null</code> if block is + * <code>false</code> and there is no data to be processed. + * + * @throws IOException If a problem occurs reading the data for the frame. + */ + public WsFrame nextFrame(boolean block) throws IOException { + frame = WsFrame.nextFrame(event, block); + if (frame != null) { + readThisFragment = 0; + remaining = frame.getPayLoadLength(); + } + return frame; + } + + + // ----------------------------------------------------- InputStream methods + + @Override + public int read() throws IOException { + + makePayloadDataAvailable(); + + if (remaining == 0) { + return -1; + } + + remaining--; + readThisFragment++; + + int masked = event.getHttpServletRequest().getInputStream().read(); + if(masked == -1) { + return -1; + } + return masked ^ + (frame.getMask()[(int) ((readThisFragment - 1) % 4)] & 0xFF); + } + + + @Override + public int read(byte b[], int off, int len) throws IOException { + + makePayloadDataAvailable(); + + if (remaining == 0) { + return -1; + } + + if (len > remaining) { + len = (int) remaining; + } + int result = event.getHttpServletRequest().getInputStream().read(b, off, len); + if(result == -1) { + return -1; + } + + for (int i = off; i < off + result; i++) { + b[i] = (byte) (b[i] ^ + frame.getMask()[(int) ((readThisFragment + i - off) % 4)]); + } + remaining -= result; + readThisFragment += result; + return result; + } + + + /* + * Ensures that there is payload data ready to read. + */ + private void makePayloadDataAvailable() throws IOException { + if (error != null) { + throw new IOException(error); + } + while (remaining == 0 && !frame.getFin()) { + // Need more data - process next frame + nextFrame(true); + while (frame.isControl()) { + if (frame.getOpCode() == Constants.OPCODE_PING) { + outbound.pong(frame.getPayLoad()); + } else if (frame.getOpCode() == Constants.OPCODE_PONG) { + // NO-OP. Swallow it. + } else if (frame.getOpCode() == Constants.OPCODE_CLOSE) { + outbound.close(frame); + } else{ + throw new IOException(MESSAGES.frameUnknownOpcode(Byte.valueOf(frame.getOpCode()))); + } + nextFrame(true); + } + if (frame.getOpCode() != Constants.OPCODE_CONTINUATION) { + error = MESSAGES.frameNotContinuation(Byte.valueOf(frame.getOpCode())); + throw new IOException(error); + } + } + } +} Added: branches/7.2.x/src/main/java/org/apache/catalina/websocket/WsOutbound.java =================================================================== --- branches/7.2.x/src/main/java/org/apache/catalina/websocket/WsOutbound.java (rev 0) +++ branches/7.2.x/src/main/java/org/apache/catalina/websocket/WsOutbound.java 2013-06-17 14:08:53 UTC (rev 2223) @@ -0,0 +1,454 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.catalina.websocket; + +import static org.jboss.web.CatalinaMessages.MESSAGES; + +import java.io.IOException; +import java.nio.ByteBuffer; +import java.nio.CharBuffer; +import java.nio.charset.CharsetEncoder; +import java.nio.charset.CoderResult; + +import javax.servlet.ServletOutputStream; + +import org.apache.tomcat.util.buf.EncodingToCharset; +import org.jboss.servlet.http.HttpEvent; + +/** + * Provides the means to write WebSocket messages to the client. All methods + * that write to the client (or update a buffer that is later written to the + * client) are synchronized to prevent multiple threads trying to write to the + * client at the same time. + */ +public class WsOutbound { + + public static final int DEFAULT_BUFFER_SIZE = 8192; + + private HttpEvent event; + private StreamInbound streamInbound; + private ByteBuffer bb; + private CharBuffer cb; + private boolean closed = false; + private Boolean text = null; + private boolean firstFrame = true; + + + public WsOutbound(HttpEvent event, + StreamInbound streamInbound) { + this(event, streamInbound, DEFAULT_BUFFER_SIZE, + DEFAULT_BUFFER_SIZE); + } + + + public WsOutbound(HttpEvent event, StreamInbound streamInbound, + int byteBufferSize, int charBufferSize) { + this.event = event; + this.streamInbound = streamInbound; + this.bb = ByteBuffer.allocate(byteBufferSize); + this.cb = CharBuffer.allocate(charBufferSize); + } + + + /** + * Adds the data to the buffer for binary data. If a textual message is + * currently in progress that message will be completed and a new binary + * message started. If the buffer for binary data is full, the buffer will + * be flushed and a new binary continuation fragment started. + * + * @param b The byte (only the least significant byte is used) of data to + * send to the client. + * + * @throws IOException If a flush is required and an error occurs writing + * the WebSocket frame to the client + */ + public synchronized void writeBinaryData(int b) throws IOException { + if (closed) { + throw new IOException(MESSAGES.outboundClosed()); + } + + if (bb.position() == bb.capacity()) { + doFlush(false); + } + if (text == null) { + text = Boolean.FALSE; + } else if (text == Boolean.TRUE) { + // Flush the character data + flush(); + text = Boolean.FALSE; + } + bb.put((byte) (b & 0xFF)); + } + + + /** + * Adds the data to the buffer for textual data. If a binary message is + * currently in progress that message will be completed and a new textual + * message started. If the buffer for textual data is full, the buffer will + * be flushed and a new textual continuation fragment started. + * + * @param c The character to send to the client. + * + * @throws IOException If a flush is required and an error occurs writing + * the WebSocket frame to the client + */ + public synchronized void writeTextData(char c) throws IOException { + if (closed) { + throw new IOException(MESSAGES.outboundClosed()); + } + + if (cb.position() == cb.capacity()) { + doFlush(false); + } + + if (text == null) { + text = Boolean.TRUE; + } else if (text == Boolean.FALSE) { + // Flush the binary data + flush(); + text = Boolean.TRUE; + } + cb.append(c); + } + + + /** + * Flush any message (binary or textual) that may be buffered and then send + * a WebSocket binary message as a single frame with the provided buffer as + * the payload of the message. + * + * @param msgBb The buffer containing the payload + * + * @throws IOException If an error occurs writing to the client + */ + public synchronized void writeBinaryMessage(ByteBuffer msgBb) + throws IOException { + + if (closed) { + throw new IOException(MESSAGES.outboundClosed()); + } + + if (text != null) { + // Empty the buffer + flush(); + } + text = Boolean.FALSE; + doWriteBytes(msgBb, true); + } + + + /** + * Flush any message (binary or textual) that may be buffered and then send + * a WebSocket text message as a single frame with the provided buffer as + * the payload of the message. + * + * @param msgCb The buffer containing the payload + * + * @throws IOException If an error occurs writing to the client + */ + public synchronized void writeTextMessage(CharBuffer msgCb) + throws IOException { + + if (closed) { + throw new IOException(MESSAGES.outboundClosed()); + } + + if (text != null) { + // Empty the buffer + flush(); + } + text = Boolean.TRUE; + doWriteText(msgCb, true); + } + + + /** + * Flush any message (binary or textual) that may be buffered. + * + * @throws IOException If an error occurs writing to the client + */ + public synchronized void flush() throws IOException { + if (closed) { + throw new IOException(MESSAGES.outboundClosed()); + } + doFlush(true); + } + + + private void doFlush(boolean finalFragment) throws IOException { + if (text == null) { + // No data + return; + } + if (text.booleanValue()) { + cb.flip(); + doWriteText(cb, finalFragment); + } else { + bb.flip(); + doWriteBytes(bb, finalFragment); + } + } + + + /** + * Respond to a client close by sending a close that echoes the status code + * and message. + * + * @param frame The close frame received from a client + * + * @throws IOException If an error occurs writing to the client + */ + protected void close(WsFrame frame) throws IOException { + if (frame.getPayLoadLength() > 0) { + // Must be status (2 bytes) plus optional message + if (frame.getPayLoadLength() == 1) { + throw new IOException(); + } + int status = (frame.getPayLoad().get() & 0xFF) << 8; + status += frame.getPayLoad().get() & 0xFF; + + if (validateCloseStatus(status)) { + // Echo the status back to the client + close(status, frame.getPayLoad()); + } else { + // Invalid close code + close(Constants.STATUS_PROTOCOL_ERROR, null); + } + } else { + // No status + close(0, null); + } + } + + + private boolean validateCloseStatus(int status) { + + if (status == Constants.STATUS_CLOSE_NORMAL || + status == Constants.STATUS_SHUTDOWN || + status == Constants.STATUS_PROTOCOL_ERROR || + status == Constants.STATUS_UNEXPECTED_DATA_TYPE || + status == Constants.STATUS_BAD_DATA || + status == Constants.STATUS_POLICY_VIOLATION || + status == Constants.STATUS_MESSAGE_TOO_LARGE || + status == Constants.STATUS_REQUIRED_EXTENSION || + status == Constants.STATUS_UNEXPECTED_CONDITION || + (status > 2999 && status < 5000)) { + // Other 1xxx reserved / not permitted + // 2xxx reserved + // 3xxx framework defined + // 4xxx application defined + return true; + } + // <1000 unused + // >4999 undefined + return false; + } + + + /** + * Send a close message to the client + * + * @param status Must be a valid status code or zero to send no code + * @param data Optional message. If message is defined, a valid status + * code must be provided. + * + * @throws IOException If an error occurs writing to the client + */ + public synchronized void close(int status, ByteBuffer data) + throws IOException { + + if (closed) { + return; + } + + // Send any partial data we have + try { + doFlush(false); + } finally { + closed = true; + } + + ServletOutputStream os = event.getHttpServletResponse().getOutputStream(); + os.write(0x88); + if (status == 0) { + os.write(0); + } else if (data == null || data.position() == data.limit()) { + os.write(2); + os.write(status >>> 8); + os.write(status); + } else { + os.write(2 + data.limit() - data.position()); + os.write(status >>> 8); + os.write(status); + os.write(data.array(), data.position(), + data.limit() - data.position()); + } + os.flush(); + + bb = null; + cb = null; + } + + + /** + * Send a pong message to the client + * + * @param data Optional message. + * + * @throws IOException If an error occurs writing to the client + */ + public synchronized void pong(ByteBuffer data) throws IOException { + sendControlMessage(data, Constants.OPCODE_PONG); + } + + /** + * Send a ping message to the client + * + * @param data Optional message. + * + * @throws IOException If an error occurs writing to the client + */ + public synchronized void ping(ByteBuffer data) throws IOException { + sendControlMessage(data, Constants.OPCODE_PING); + } + + /** + * Generic function to send either a ping or a pong. + * + * @param data Optional message. + * @param opcode The byte to include as the opcode. + * + * @throws IOException If an error occurs writing to the client + */ + private synchronized void sendControlMessage(ByteBuffer data, byte opcode) throws IOException { + + if (closed) { + throw new IOException(MESSAGES.outboundClosed()); + } + + doFlush(false); + + ServletOutputStream os = event.getHttpServletResponse().getOutputStream(); + os.write(0x80 | opcode); + if (data == null) { + os.write(0); + } else { + os.write(data.limit() - data.position()); + os.write(data.array(), data.position(), + data.limit() - data.position()); + } + + os.flush(); + } + + + /** + * Writes the provided bytes as the payload in a new WebSocket frame. + * + * @param buffer The bytes to include in the payload. + * @param finalFragment Do these bytes represent the final fragment of a + * WebSocket message? + * @throws IOException + */ + private void doWriteBytes(ByteBuffer buffer, boolean finalFragment) + throws IOException { + + if (closed) { + throw new IOException(MESSAGES.outboundClosed()); + } + + try { + // Work out the first byte + int first = 0x00; + if (finalFragment) { + first = first + 0x80; + } + if (firstFrame) { + if (text.booleanValue()) { + first = first + 0x1; + } else { + first = first + 0x2; + } + } + ServletOutputStream os = event.getHttpServletResponse().getOutputStream(); + // Continuation frame is OpCode 0 + os.write(first); + + if (buffer.limit() < 126) { + os.write(buffer.limit()); + } else if (buffer.limit() < 65536) { + os.write(126); + os.write(buffer.limit() >>> 8); + os.write(buffer.limit() & 0xFF); + } else { + // Will never be more than 2^31-1 + os.write(127); + os.write(0); + os.write(0); + os.write(0); + os.write(0); + os.write(buffer.limit() >>> 24); + os.write(buffer.limit() >>> 16); + os.write(buffer.limit() >>> 8); + os.write(buffer.limit() & 0xFF); + } + + // Write the content + os.write(buffer.array(), buffer.arrayOffset(), + buffer.limit()); + os.flush(); + + // Reset + if (finalFragment) { + text = null; + firstFrame = true; + } else { + firstFrame = false; + } + bb.clear(); + } catch (IOException ioe) { + // Any IOException is terminal. Make sure the Inbound side knows + // that something went wrong. + streamInbound.doOnClose(Constants.STATUS_CLOSED_UNEXPECTEDLY); + throw ioe; + } + } + + + /* + * Convert the textual message to bytes and then output it. + */ + private void doWriteText(CharBuffer buffer, boolean finalFragment) + throws IOException { + CharsetEncoder encoder = EncodingToCharset.UTF_8.newEncoder(); + do { + CoderResult cr = encoder.encode(buffer, bb, true); + if (cr.isError()) { + cr.throwException(); + } + bb.flip(); + if (buffer.hasRemaining()) { + doWriteBytes(bb, false); + } else { + doWriteBytes(bb, finalFragment); + } + } while (buffer.hasRemaining()); + + // Reset - bb will be cleared in doWriteBytes() + cb.clear(); + } +} Added: branches/7.2.x/src/main/java/org/apache/tomcat/util/buf/Utf8Decoder.java =================================================================== --- branches/7.2.x/src/main/java/org/apache/tomcat/util/buf/Utf8Decoder.java (rev 0) +++ branches/7.2.x/src/main/java/org/apache/tomcat/util/buf/Utf8Decoder.java 2013-06-17 14:08:53 UTC (rev 2223) @@ -0,0 +1,293 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.tomcat.util.buf; + +import java.nio.ByteBuffer; +import java.nio.CharBuffer; +import java.nio.charset.CharsetDecoder; +import java.nio.charset.CoderResult; + +/** + * Decodes bytes to UTF-8. Extracted from Apache Harmony and modified to reject + * code points from U+D800 to U+DFFF as per RFC3629. The standard Java decoder + * does not reject these. It has also been modified to reject code points + * greater than U+10FFFF which the standard Java decoder rejects but the harmony + * one does not. + */ +public class Utf8Decoder extends CharsetDecoder { + + // The next table contains information about UTF-8 charset and + // correspondence of 1st byte to the length of sequence + // For information please visit http://www.ietf.org/rfc/rfc3629.txt + // + // Please note, o means 0, actually. + // ------------------------------------------------------------------- + // 0 1 2 3 Value + // ------------------------------------------------------------------- + // oxxxxxxx 00000000 00000000 0xxxxxxx + // 11oyyyyy 1oxxxxxx 00000000 00000yyy yyxxxxxx + // 111ozzzz 1oyyyyyy 1oxxxxxx 00000000 zzzzyyyy yyxxxxxx + // 1111ouuu 1ouuzzzz 1oyyyyyy 1oxxxxxx 000uuuuu zzzzyyyy yyxxxxxx + private static final int remainingBytes[] = { + // 1owwwwww + -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, + -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, + -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, + -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, + // 11oyyyyy + -1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, + 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, + // 111ozzzz + 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, + // 1111ouuu + 3, 3, 3, 3, 3, -1, -1, -1, + // > 11110111 + -1, -1, -1, -1, -1, -1, -1, -1}; + private static final int remainingNumbers[] = {0, // 0 1 2 3 + 4224, // (01o00000b << 6)+(1o000000b) + 401536, // (011o0000b << 12)+(1o000000b << 6)+(1o000000b) + 29892736 // (0111o000b << 18)+(1o000000b << 12)+(1o000000b << + // 6)+(1o000000b) + }; + private static final int lowerEncodingLimit[] = {-1, 0x80, 0x800, 0x10000}; + + + public Utf8Decoder() { + super(EncodingToCharset.UTF_8, 1.0f, 1.0f); + } + + + @Override + protected CoderResult decodeLoop(ByteBuffer in, CharBuffer out) { + if (in.hasArray() && out.hasArray()) { + return decodeHasArray(in, out); + } + return decodeNotHasArray(in, out); + } + + + private CoderResult decodeNotHasArray(ByteBuffer in, CharBuffer out) { + int outRemaining = out.remaining(); + int pos = in.position(); + int limit = in.limit(); + try { + while (pos < limit) { + if (outRemaining == 0) { + return CoderResult.OVERFLOW; + } + int jchar = in.get(); + if (jchar < 0) { + jchar = jchar & 0x7F; + int tail = remainingBytes[jchar]; + if (tail == -1) { + return CoderResult.malformedForLength(1); + } + if (limit - pos < 1 + tail) { + // No early test for invalid sequences here as peeking + // at the next byte is harder + return CoderResult.UNDERFLOW; + } + int nextByte; + for (int i = 0; i < tail; i++) { + nextByte = in.get() & 0xFF; + if ((nextByte & 0xC0) != 0x80) { + return CoderResult.malformedForLength(1 + i); + } + jchar = (jchar << 6) + nextByte; + } + jchar -= remainingNumbers[tail]; + if (jchar < lowerEncodingLimit[tail]) { + // Should have been encoded in a fewer octets + return CoderResult.malformedForLength(1); + } + pos += tail; + } + // Apache Tomcat added test + if (jchar >= 0xD800 && jchar <= 0xDFFF) { + return CoderResult.unmappableForLength(3); + } + // Apache Tomcat added test + if (jchar > 0x10FFFF) { + return CoderResult.unmappableForLength(4); + } + if (jchar <= 0xffff) { + out.put((char) jchar); + outRemaining--; + } else { + if (outRemaining < 2) { + return CoderResult.OVERFLOW; + } + out.put((char) ((jchar >> 0xA) + 0xD7C0)); + out.put((char) ((jchar & 0x3FF) + 0xDC00)); + outRemaining -= 2; + } + pos++; + } + return CoderResult.UNDERFLOW; + } finally { + in.position(pos); + } + } + + + private CoderResult decodeHasArray(ByteBuffer in, CharBuffer out) { + int outRemaining = out.remaining(); + int pos = in.position(); + int limit = in.limit(); + final byte[] bArr = in.array(); + final char[] cArr = out.array(); + final int inIndexLimit = limit + in.arrayOffset(); + int inIndex = pos + in.arrayOffset(); + int outIndex = out.position() + out.arrayOffset(); + // if someone would change the limit in process, + // he would face consequences + for (; inIndex < inIndexLimit && outRemaining > 0; inIndex++) { + int jchar = bArr[inIndex]; + if (jchar < 0) { + jchar = jchar & 0x7F; + // If first byte is invalid, tail will be set to -1 + int tail = remainingBytes[jchar]; + if (tail == -1) { + in.position(inIndex - in.arrayOffset()); + out.position(outIndex - out.arrayOffset()); + return CoderResult.malformedForLength(1); + } + // Additional checks to detect invalid sequences ASAP + // Checks derived from Unicode 6.2, Chapter 3, Table 3-7 + // Check 2nd byte + int tailAvailable = inIndexLimit - inIndex - 1; + if (tailAvailable > 0) { + // First byte C2..DF, second byte 80..BF + if (jchar > 0x41 && jchar < 0x60 && + (bArr[inIndex + 1] & 0xC0) != 0x80) { + in.position(inIndex - in.arrayOffset()); + out.position(outIndex - out.arrayOffset()); + return CoderResult.malformedForLength(1); + } + // First byte E0, second byte A0..BF + if (jchar == 0x60 && (bArr[inIndex + 1] & 0xE0) != 0xA0) { + in.position(inIndex - in.arrayOffset()); + out.position(outIndex - out.arrayOffset()); + return CoderResult.malformedForLength(1); + } + // First byte E1..EC, second byte 80..BF + if (jchar > 0x60 && jchar < 0x6D && + (bArr[inIndex + 1] & 0xC0) != 0x80) { + in.position(inIndex - in.arrayOffset()); + out.position(outIndex - out.arrayOffset()); + return CoderResult.malformedForLength(1); + } + // First byte ED, second byte 80..9F + if (jchar == 0x6D && (bArr[inIndex + 1] & 0xE0) != 0x80) { + in.position(inIndex - in.arrayOffset()); + out.position(outIndex - out.arrayOffset()); + return CoderResult.malformedForLength(1); + } + // First byte EE..EF, second byte 80..BF + if (jchar > 0x6D && jchar < 0x70 && + (bArr[inIndex + 1] & 0xC0) != 0x80) { + in.position(inIndex - in.arrayOffset()); + out.position(outIndex - out.arrayOffset()); + return CoderResult.malformedForLength(1); + } + // First byte F0, second byte 90..BF + if (jchar == 0x70 && + ((bArr[inIndex + 1] & 0xFF) < 0x90 || + (bArr[inIndex + 1] & 0xFF) > 0xBF)) { + in.position(inIndex - in.arrayOffset()); + out.position(outIndex - out.arrayOffset()); + return CoderResult.malformedForLength(1); + } + // First byte F1..F3, second byte 80..BF + if (jchar > 0x70 && jchar < 0x74 && + (bArr[inIndex + 1] & 0xC0) != 0x80) { + in.position(inIndex - in.arrayOffset()); + out.position(outIndex - out.arrayOffset()); + return CoderResult.malformedForLength(1); + } + // First byte F4, second byte 80..8F + if (jchar == 0x74 && + (bArr[inIndex + 1] & 0xF0) != 0x80) { + in.position(inIndex - in.arrayOffset()); + out.position(outIndex - out.arrayOffset()); + return CoderResult.malformedForLength(1); + } + } + // Check third byte if present and expected + if (tailAvailable > 1 && tail > 1) { + if ((bArr[inIndex + 2] & 0xC0) != 0x80) { + in.position(inIndex - in.arrayOffset()); + out.position(outIndex - out.arrayOffset()); + return CoderResult.malformedForLength(2); + } + } + // Check fourth byte if present and expected + if (tailAvailable > 2 && tail > 2) { + if ((bArr[inIndex + 3] & 0xC0) != 0x80) { + in.position(inIndex - in.arrayOffset()); + out.position(outIndex - out.arrayOffset()); + return CoderResult.malformedForLength(3); + } + } + if (tailAvailable < tail) { + break; + } + for (int i = 0; i < tail; i++) { + int nextByte = bArr[inIndex + i + 1] & 0xFF; + if ((nextByte & 0xC0) != 0x80) { + in.position(inIndex - in.arrayOffset()); + out.position(outIndex - out.arrayOffset()); + return CoderResult.malformedForLength(1 + i); + } + jchar = (jchar << 6) + nextByte; + } + jchar -= remainingNumbers[tail]; + if (jchar < lowerEncodingLimit[tail]) { + // Should have been encoded in fewer octets + in.position(inIndex - in.arrayOffset()); + out.position(outIndex - out.arrayOffset()); + return CoderResult.malformedForLength(1); + } + inIndex += tail; + } + // Apache Tomcat added test + if (jchar >= 0xD800 && jchar <= 0xDFFF) { + return CoderResult.unmappableForLength(3); + } + // Apache Tomcat added test + if (jchar > 0x10FFFF) { + return CoderResult.unmappableForLength(4); + } + if (jchar <= 0xffff) { + cArr[outIndex++] = (char) jchar; + outRemaining--; + } else { + if (outRemaining < 2) { + return CoderResult.OVERFLOW; + } + cArr[outIndex++] = (char) ((jchar >> 0xA) + 0xD7C0); + cArr[outIndex++] = (char) ((jchar & 0x3FF) + 0xDC00); + outRemaining -= 2; + } + } + in.position(inIndex - in.arrayOffset()); + out.position(outIndex - out.arrayOffset()); + return (outRemaining == 0 && inIndex < inIndexLimit) ? + CoderResult.OVERFLOW : + CoderResult.UNDERFLOW; + } +} Modified: branches/7.2.x/src/main/java/org/jboss/web/CatalinaMessages.java =================================================================== --- branches/7.2.x/src/main/java/org/jboss/web/CatalinaMessages.java 2013-06-16 05:50:51 UTC (rev 2222) +++ branches/7.2.x/src/main/java/org/jboss/web/CatalinaMessages.java 2013-06-17 14:08:53 UTC (rev 2223) @@ -972,4 +972,34 @@ @Message(id = 359, value = "Exception releasing filter %s") String errorStoppingFilter(String filterName); + @Message(id = 360, value = "Unable to cast to the internal request class in order to complete HTTP upgrade") + String errorUpgrading(); + + @Message(id = 361, value = "The buffer is not big enough to contain the message currently being processed") + String bufferTooSmall(); + + @Message(id = 362, value = "The client frame was not masked but all client frames must be masked") + String frameNotMasked(); + + @Message(id = 363, value = "A sequence of bytes was received that did not represent valid UTF-8") + String frameInvalidUtf8(); + + @Message(id = 364, value = "The end of the stream was reached when trying to read the first byte of a new WebSocket frame") + String frameEos(); + + @Message(id = 365, value = "Failed to read the first byte of the next WebSocket frame. The return value from the read was [%s]") + String frameFailedRead(int value); + + @Message(id = 366, value = "An attempt was made to access the request object passed to WebSocketServlet.createWebSocketInbound() outside of that method") + IllegalStateException invalidWrapper(); + + @Message(id = 367, value = "A frame with the unrecognized OpCode [%s] was received") + String frameUnknownOpcode(byte opcode); + + @Message(id = 368, value = "A frame with the OpCode [%s] was received when a continuation frame was expected") + String frameNotContinuation(byte opcode); + + @Message(id = 369, value = "The WebSocket connection has been closed") + String outboundClosed(); + }