t push origin atmosphere-2.4.xMerge branch 'shanielh-atmosphere-2.4.x' into atmosphere-2.4.x

Author: jfarcand

modules/cpr/pom.xml

@@ -91,7 +91,7 @@
         <dependency>
             <groupId>org.eclipse.jetty.websocket</groupId>
             <artifactId>websocket-server</artifactId>
-            <version>${jetty9-version}</version>
+            <version>${jetty9_3-version}</version>
             <scope>provided</scope>
             <optional>true</optional>
             <exclusions>

modules/cpr/src/main/java/org/atmosphere/cache/UUIDBroadcasterCache.java

@@ -24,6 +24,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.io.Serializable;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.HashSet;
@@ -52,7 +53,7 @@ public class UUIDBroadcasterCache implements BroadcasterCache {
 
     private final static Logger logger = LoggerFactory.getLogger(UUIDBroadcasterCache.class);
 
-    private final Map<String, ConcurrentLinkedQueue> messages = new ConcurrentHashMap<String, ConcurrentLinkedQueue>();
+    private final Map<String, ClientQueue> messages = new ConcurrentHashMap<String, ClientQueue>();
     private final Map<String, Long> activeClients = new ConcurrentHashMap<String, Long>();
     protected final List<BroadcasterCacheInspector> inspectors = new LinkedList<BroadcasterCacheInspector>();
     private ScheduledFuture scheduledFuture;
@@ -64,6 +65,30 @@ public class UUIDBroadcasterCache implements BroadcasterCache {
     protected final List<BroadcasterCacheListener> listeners = new LinkedList<BroadcasterCacheListener>();
     private UUIDProvider uuidProvider;
 
+    /**
+     * This class wraps all messages to be delivered to a client. The class is thread safe to be accessed in a
+     * concurrent context.
+     */
+    public final static class ClientQueue implements Serializable {
+        private static final long serialVersionUID = -126253550299206646L;
+
+        private final ConcurrentLinkedQueue<CacheMessage> queue = new ConcurrentLinkedQueue<CacheMessage>();
+        private final Set<String> ids = Collections.synchronizedSet(new HashSet<String>());
+
+        public ConcurrentLinkedQueue<CacheMessage> getQueue() {
+            return queue;
+        }
+
+        public Set<String> getIds() {
+            return ids;
+        }
+
+        @Override
+        public String toString() {
+            return queue.toString();
+        }
+    }
+
     @Override
     public void configure(AtmosphereConfig config) {
         Object o = config.properties().get("shared");
@@ -131,7 +156,7 @@ public CacheMessage addToCache(String broadcasterId, String uuid, BroadcastMessa
             cache = false;
         }
 
-        CacheMessage cacheMessage = new CacheMessage(messageId, message.message(), uuid);
+        CacheMessage cacheMessage = new CacheMessage(messageId, message.message(), uuid);;
         if (cache) {
             if (uuid.equals(NULL)) {
                 //no clients are connected right now, caching message for all active clients
@@ -149,33 +174,38 @@ public CacheMessage addToCache(String broadcasterId, String uuid, BroadcastMessa
     @Override
     public List<Object> retrieveFromCache(String broadcasterId, String uuid) {
 
-        ConcurrentLinkedQueue<CacheMessage> clientQueue = messages.get(uuid);
-        CacheMessage message;
         List<Object> result = new ArrayList<Object>();
-        if (clientQueue == null) {
-            logger.debug("client queue is null (not yet created), hence returning back for broadcaster %s and uuid %s",
-                    broadcasterId, uuid);
-            return result;
+
+        ClientQueue clientQueue;
+        cacheCandidate(broadcasterId, uuid);
+        clientQueue = messages.remove(uuid);
+        ConcurrentLinkedQueue<CacheMessage> clientMessages;
+        if (clientQueue != null) {
+            clientMessages = clientQueue.getQueue();
+
+            for (CacheMessage cacheMessage : clientMessages) {
+                result.add(cacheMessage.getMessage());
+            }
         }
 
-        while ((message = clientQueue.poll()) != null) {
-            result.add(message);
+        if (logger.isTraceEnabled()) {
+            logger.trace("Retrieved for AtmosphereResource {} cached messages {}", uuid, result);
+            logger.trace("Available cached message {}", messages);
         }
 
         return result;
     }
 
     @Override
     public BroadcasterCache clearCache(String broadcasterId, String uuid, CacheMessage message) {
-        ConcurrentLinkedQueue<CacheMessage> clientQueue = messages.remove(uuid);
-
-        if (clientQueue == null) {
-            logger.error("Invalid State, no Queue available");
-            return this;
+        ClientQueue clientQueue;
+        clientQueue = messages.get(uuid);
+        if (clientQueue != null) {
+            logger.trace("Removing for AtmosphereResource {} cached message {}", uuid, message.getMessage());
+            notifyRemoveCache(broadcasterId, message);
+            clientQueue.getQueue().remove(message);
+            clientQueue.getIds().remove(message.getId());
         }
-
-        logger.trace("Removing for AtmosphereResource {} cached message {}", uuid, message.getMessage());
-        notifyRemoveCache(broadcasterId, message);
         return this;
     }
 
@@ -210,22 +240,21 @@ private void addMessageIfNotExists(String broadcasterId, String clientId, CacheM
     }
 
     private void addMessage(String broadcasterId, String clientId, CacheMessage message) {
-        ConcurrentLinkedQueue clientQueue = messages.get(clientId);
+        ClientQueue clientQueue = messages.get(clientId);
         if (clientQueue == null) {
-            synchronized (message) {
-                clientQueue = new ConcurrentLinkedQueue();
-                // Make sure the client is not in the process of being invalidated
-                if (activeClients.get(clientId) != null) {
-                    messages.put(clientId, clientQueue);
-                } else {
-                    // The entry has been invalidated
-                    logger.debug("Client {} is no longer active. Not caching message {}}", clientId, message);
-                    return;
-                }
+            clientQueue = new ClientQueue();
+            // Make sure the client is not in the process of being invalidated
+            if (activeClients.get(clientId) != null) {
+                messages.put(clientId, clientQueue);
+            } else {
+                // The entry has been invalidated
+                logger.debug("Client {} is no longer active. Not caching message {}}", clientId, message);
+                return;
             }
         }
         notifyAddCache(broadcasterId, message);
-        messages.put(clientId, clientQueue);
+        clientQueue.getQueue().offer(message);
+        clientQueue.getIds().add(message.getId());
     }
 
     private void notifyAddCache(String broadcasterId, CacheMessage message) {
@@ -249,11 +278,11 @@ private void notifyRemoveCache(String broadcasterId, CacheMessage message) {
     }
 
     private boolean hasMessage(String clientId, String messageId) {
-        ConcurrentLinkedQueue clientQueue = messages.get(clientId);
-        return clientQueue != null && clientQueue.contains(messageId);
+        ClientQueue clientQueue = messages.get(clientId);
+        return clientQueue != null && clientQueue.getIds().contains(messageId);
     }
 
-    public Map<String, ConcurrentLinkedQueue> messages() {
+    public Map<String, ClientQueue> messages() {
         return messages;
     }
 
@@ -296,7 +325,7 @@ protected void invalidateExpiredEntries() {
 
         for (String msg : messages().keySet()) {
             if (!activeClients().containsKey(msg)) {
-                messages.remove(msg);
+                messages().remove(msg);
             }
         }
     }

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

@@ -0,0 +1,193 @@
+/*
+ * Copyright 2017 Async-IO.org
+ *
+ * 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.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.ServletUpgradeRequest;
+import org.eclipse.jetty.websocket.servlet.ServletUpgradeResponse;
+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 java.lang.reflect.Method;
+
+abstract class AbstractJetty9AsyncSupportWithWebSocket extends Servlet30CometSupport {
+
+    private static final Logger staticLogger = LoggerFactory.getLogger(AbstractJetty9AsyncSupportWithWebSocket.class);
+
+    private final Logger logger;
+
+
+    static WebSocketPolicy GetPolicy(final AtmosphereConfig 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));
+        }
+
+        try {
+            // Crazy Jetty API Incompatibility
+            String serverInfo = config.getServletConfig().getServletContext().getServerInfo();
+            boolean isJetty91Plus = false;
+            if (serverInfo != null) {
+                int version = Integer.valueOf(serverInfo.split("/")[1].substring(0, 3).replace(".", ""));
+                isJetty91Plus = version > 90;
+            }
+
+            max = config.getInitParameter(ApplicationConfig.WEBSOCKET_MAXTEXTSIZE);
+            if (max != null) {
+                //policy.setMaxMessageSize(Integer.parseInt(max));
+                Method m;
+                if (isJetty91Plus) {
+                    m = policy.getClass().getMethod("setMaxTextMessageSize", int.class);
+                } else {
+                    m = policy.getClass().getMethod("setMaxMessageSize", long.class);
+                }
+                m.invoke(policy, Integer.parseInt(max));
+            }
+
+            max = config.getInitParameter(ApplicationConfig.WEBSOCKET_MAXBINARYSIZE);
+            if (max != null) {
+                //policy.setMaxMessageSize(Integer.parseInt(max));
+                Method m;
+                if (isJetty91Plus) {
+                    m = policy.getClass().getMethod("setMaxBinaryMessageSize", int.class);
+                } else {
+                    m = policy.getClass().getMethod("setMaxMessageSize", long.class);
+                }
+                m.invoke(policy, Integer.parseInt(max));
+            }
+        } catch (Exception ex) {
+            staticLogger.warn("", ex);
+        }
+
+        return policy;
+    }
+
+    AbstractJetty9AsyncSupportWithWebSocket(AtmosphereConfig config, Logger logger) {
+        super(config);
+        this.logger = logger;
+    }
+
+    WebSocketCreator buildCreator(final HttpServletRequest request, final HttpServletResponse response, final WebSocketProcessor webSocketProcessor) {
+        return new WebSocketCreator() {
+
+            // @Override  9.0.x
+            public Object createWebSocket(UpgradeRequest upgradeRequest, UpgradeResponse upgradeResponse) {
+
+                ServletWebSocketRequest r = ServletWebSocketRequest.class.cast(upgradeRequest);
+                r.getExtensions().clear();
+
+                if (!webSocketProcessor.handshake(request)) {
+                    try {
+                        response.sendError(HttpServletResponse.SC_FORBIDDEN, "WebSocket requests rejected.");
+                    } catch (IOException e) {
+                        logger.trace("", e);
+                    }
+                    return null;
+                }
+
+                return new Jetty9WebSocketHandler(request, config.framework(), webSocketProcessor);
+            }
+
+            // @Override 9.1.x
+            public Object createWebSocket(ServletUpgradeRequest req, ServletUpgradeResponse resp) {
+                req.getExtensions().clear();
+
+                if (!webSocketProcessor.handshake(request)) {
+                    try {
+                        response.sendError(HttpServletResponse.SC_FORBIDDEN, "WebSocket requests rejected.");
+                    } catch (IOException e) {
+                        logger.trace("", e);
+                    }
+                    return null;
+                }
+                return new Jetty9WebSocketHandler(request, config.framework(), webSocketProcessor);
+            }
+        };
+    }
+
+    abstract WebSocketServerFactory getWebSocketFactory();
+
+    @Override
+    public Action service(AtmosphereRequest req, AtmosphereResponse res)
+            throws IOException, ServletException {
+
+        Action action;
+        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) {
+                WebSocket.notSupported(req, res);
+                return Action.CANCELLED;
+            } else {
+                return super.service(req, res);
+            }
+        } else {
+            if (getWebSocketFactory() != null && !b) {
+                req.setAttribute(WebSocket.WEBSOCKET_INITIATED, true);
+                getWebSocketFactory().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;
+    }
+
+}