Fix for #952

modules/cpr/pom.xml

@@ -73,6 +73,32 @@
             <scope>provided</scope>
             <optional>true</optional>
         </dependency>
+        <dependency>
+            <groupId>org.eclipse.jetty.websocket</groupId>
+            <artifactId>websocket-server</artifactId>
+            <version>${jetty9-version}</version>
+            <scope>provided</scope>
+            <optional>true</optional>
+            <exclusions>
+                <exclusion>
+                    <groupId>javax.servlet</groupId>
+                    <artifactId>servlet-api</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
+        <dependency>
+            <groupId>org.eclipse.jetty</groupId>
+            <artifactId>jetty-util</artifactId>
+            <version>${jetty9-version}</version>
+            <scope>provided</scope>
+            <optional>true</optional>
+            <exclusions>
+                <exclusion>
+                    <groupId>javax.servlet</groupId>
+                    <artifactId>servlet-api</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
         <dependency>
             <groupId>org.eclipse.jetty</groupId>
             <artifactId>jetty-continuation</artifactId>

modules/cpr/src/main/java/org/atmosphere/container/Jetty9AsyncSupportWithWebSocket.java

@@ -0,0 +1,157 @@
+/*
+* Copyright 2013 Jeanfrancois Arcand
+*
+* Licensed 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.atmosphere.container;
+
+import org.atmosphere.cpr.Action;
+import org.atmosphere.cpr.ApplicationConfig;
+import org.atmosphere.cpr.AtmosphereConfig;
+import org.atmosphere.cpr.AtmosphereRequest;
+import org.atmosphere.cpr.AtmosphereResource;
+import org.atmosphere.cpr.AtmosphereResponse;
+import org.atmosphere.cpr.WebSocketProcessorFactory;
+import org.atmosphere.util.Utils;
+import org.atmosphere.websocket.WebSocket;
+import org.atmosphere.websocket.WebSocketProcessor;
+import org.eclipse.jetty.websocket.api.UpgradeRequest;
+import org.eclipse.jetty.websocket.api.UpgradeResponse;
+import org.eclipse.jetty.websocket.api.WebSocketBehavior;
+import org.eclipse.jetty.websocket.api.WebSocketPolicy;
+import org.eclipse.jetty.websocket.server.ServletWebSocketRequest;
+import org.eclipse.jetty.websocket.server.WebSocketServerFactory;
+import org.eclipse.jetty.websocket.servlet.WebSocketCreator;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.servlet.ServletException;
+import javax.servlet.http.HttpServletRequest;
+import javax.servlet.http.HttpServletResponse;
+import java.io.IOException;
+
+import static org.atmosphere.cpr.HeaderConfig.X_ATMOSPHERE_ERROR;
+
+/**
+ * Jetty 9 WebSocket support.
+ *
+ * @author Jeanfrancois Arcand
+ */
+public class Jetty9AsyncSupportWithWebSocket extends Jetty7CometSupport {
+    private static final Logger logger = LoggerFactory.getLogger(Jetty9AsyncSupportWithWebSocket.class);
+    private final WebSocketServerFactory webSocketFactory;
+
+    public Jetty9AsyncSupportWithWebSocket(final AtmosphereConfig config) {
+        super(config);
+
+        String bs = config.getInitParameter(ApplicationConfig.WEBSOCKET_BUFFER_SIZE);
+        WebSocketPolicy policy = new WebSocketPolicy(WebSocketBehavior.SERVER);
+        if (bs != null) {
+            policy.setInputBufferSize(Integer.parseInt(bs));
+        }
+
+        String max = config.getInitParameter(ApplicationConfig.WEBSOCKET_IDLETIME);
+        if (max != null) {
+            policy.setIdleTimeout(Integer.parseInt(max));
+        }
+
+        max = config.getInitParameter(ApplicationConfig.WEBSOCKET_MAXTEXTSIZE);
+        if (max != null) {
+            policy.setMaxMessageSize(Integer.parseInt(max));
+        }
+
+        max = config.getInitParameter(ApplicationConfig.WEBSOCKET_MAXBINARYSIZE);
+        if (max != null) {
+            policy.setMaxMessageSize(Integer.parseInt(max));
+        }
+        final WebSocketProcessor webSocketProcessor = WebSocketProcessorFactory.getDefault().getWebSocketProcessor(config.framework());
+
+        webSocketFactory = new WebSocketServerFactory(policy) {
+            @Override
+            public boolean acceptWebSocket(final HttpServletRequest request, HttpServletResponse response) throws IOException {
+                setCreator(new WebSocketCreator() {
+
+                    @Override
+                    public Object createWebSocket(UpgradeRequest upgradeRequest, UpgradeResponse upgradeResponse) {
+
+                        ServletWebSocketRequest r = ServletWebSocketRequest.class.cast(upgradeRequest);
+                        r.getExtensions().clear();
+
+                        return new Jetty9WebSocketHandler(request, config.framework(), webSocketProcessor);
+                    }
+                });
+
+                return super.acceptWebSocket(request, response);
+            }
+        };
+
+        try {
+            webSocketFactory.start();
+        } catch (Exception e) {
+            logger.error("", e);
+        }
+    }
+
+    /**
+     * {@inheritDoc}
+     */
+    @Override
+    public Action service(AtmosphereRequest req, AtmosphereResponse res)
+            throws IOException, ServletException {
+
+        Action action = null;
+        Boolean b = (Boolean) req.getAttribute(WebSocket.WEBSOCKET_INITIATED);
+        if (b == null) b = Boolean.FALSE;
+
+        if (!Utils.webSocketEnabled(req) && req.getAttribute(WebSocket.WEBSOCKET_ACCEPT_DONE) == null) {
+            if (req.resource() != null && req.resource().transport() == AtmosphereResource.TRANSPORT.WEBSOCKET) {
+                logger.trace("Invalid WebSocket Specification {}", req);
+
+                res.addHeader(X_ATMOSPHERE_ERROR, "Websocket protocol not supported");
+                res.sendError(501, "Websocket protocol not supported");
+                return Action.CANCELLED;
+            } else {
+                return super.service(req, res);
+            }
+        } else {
+            if (webSocketFactory != null && !b) {
+                req.setAttribute(WebSocket.WEBSOCKET_INITIATED, true);
+                webSocketFactory.acceptWebSocket(req, res);
+                req.setAttribute(WebSocket.WEBSOCKET_ACCEPT_DONE, true);
+                return new Action();
+            }
+
+            action = suspended(req, res);
+            if (action.type() == Action.TYPE.SUSPEND) {
+            } else if (action.type() == Action.TYPE.RESUME) {
+                req.setAttribute(WebSocket.WEBSOCKET_RESUME, true);
+            }
+        }
+
+        return action == null ? super.service(req, res) : action;
+    }
+
+    /**
+     * Return the container's name.
+     */
+    public String getContainerName() {
+        return config.getServletConfig().getServletContext().getServerInfo() + " with WebSocket enabled.";
+    }
+
+    @Override
+    public boolean supportWebSocket() {
+        return true;
+    }
+
+
+}

modules/cpr/src/main/java/org/atmosphere/container/Jetty9WebSocketHandler.java

@@ -0,0 +1,76 @@
+package org.atmosphere.container;
+
+import org.atmosphere.container.version.Jetty9WebSocket;
+import org.atmosphere.cpr.AtmosphereFramework;
+import org.atmosphere.cpr.AtmosphereRequest;
+import org.atmosphere.websocket.WebSocket;
+import org.atmosphere.websocket.WebSocketEventListener;
+import org.atmosphere.websocket.WebSocketProcessor;
+import org.eclipse.jetty.websocket.api.Session;
+import org.eclipse.jetty.websocket.api.WebSocketListener;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.servlet.http.HttpServletRequest;
+
+public class Jetty9WebSocketHandler implements WebSocketListener {
+
+    private static final Logger logger = LoggerFactory.getLogger(Jetty9WebSocketHandler.class);
+
+    private final AtmosphereRequest request;
+    private final AtmosphereFramework framework;
+    private final WebSocketProcessor webSocketProcessor;
+    private WebSocket webSocket;
+
+    public Jetty9WebSocketHandler(HttpServletRequest request, AtmosphereFramework framework, WebSocketProcessor webSocketProcessor) {
+        this.framework = framework;
+        this.request = cloneRequest(request);
+        this.webSocketProcessor = webSocketProcessor;
+    }
+
+    private AtmosphereRequest cloneRequest(final HttpServletRequest request) {
+        try {
+            AtmosphereRequest r = AtmosphereRequest.wrap(request);
+            return AtmosphereRequest.cloneRequest(r, false, framework.getAtmosphereConfig().isSupportSession(), false);
+        } catch (Exception ex) {
+            logger.error("", ex);
+            throw new RuntimeException("Invalid WebSocket Request");
+        }
+    }
+
+    @Override
+    public void onWebSocketBinary(byte[] data, int offset, int length) {
+        logger.trace("WebSocket.onMessage (bytes)");
+        webSocketProcessor.invokeWebSocketProtocol(webSocket, data, offset, length);
+    }
+
+    @Override
+    public void onWebSocketClose(int closeCode, String s) {
+        request.destroy();
+        webSocketProcessor.close(webSocket, closeCode);
+    }
+
+    @Override
+    public void onWebSocketConnect(Session session) {
+        logger.trace("WebSocket.onOpen.");
+        webSocket = new Jetty9WebSocket(session, framework.getAtmosphereConfig());
+        try {
+            webSocketProcessor.open(webSocket, request);
+        } catch (Exception e) {
+            logger.warn("Failed to connect to WebSocket", e);
+        }
+    }
+
+    @Override
+    public void onWebSocketError(Throwable e) {
+        logger.error("", e);
+        webSocketProcessor.notifyListener(webSocket,
+                new WebSocketEventListener.WebSocketEvent(e.getMessage(), WebSocketEventListener.WebSocketEvent.TYPE.CLOSE, webSocket));
+    }
+
+    @Override
+    public void onWebSocketText(String s) {
+        logger.trace("WebSocket.onMessage (bytes)");
+        webSocketProcessor.invokeWebSocketProtocol(webSocket, s);
+    }
+}