Merge branch 2.4.2.x

Author: jfarcand

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

@@ -24,7 +24,6 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import java.io.Serializable;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.HashSet;
@@ -53,7 +52,7 @@ public class UUIDBroadcasterCache implements BroadcasterCache {
 
     private final static Logger logger = LoggerFactory.getLogger(UUIDBroadcasterCache.class);
 
-    private final Map<String, ClientQueue> messages = new ConcurrentHashMap<String, ClientQueue>();
+    private final Map<String, ConcurrentLinkedQueue> messages = new ConcurrentHashMap<String, ConcurrentLinkedQueue>();
     private final Map<String, Long> activeClients = new ConcurrentHashMap<String, Long>();
     protected final List<BroadcasterCacheInspector> inspectors = new LinkedList<BroadcasterCacheInspector>();
     private ScheduledFuture scheduledFuture;
@@ -65,30 +64,6 @@ 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");
@@ -156,7 +131,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
@@ -174,38 +149,33 @@ 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>();
-
-        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());
-            }
+        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;
         }
 
-        if (logger.isTraceEnabled()) {
-            logger.trace("Retrieved for AtmosphereResource {} cached messages {}", uuid, result);
-            logger.trace("Available cached message {}", messages);
+        while ((message = clientQueue.poll()) != null) {
+            result.add(message);
         }
 
         return result;
     }
 
     @Override
     public BroadcasterCache clearCache(String broadcasterId, String uuid, CacheMessage message) {
-        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());
+        ConcurrentLinkedQueue<CacheMessage> clientQueue = messages.remove(uuid);
+
+        if (clientQueue == null) {
+            logger.error("Invalid State, no Queue available");
+            return this;
         }
+
+        logger.trace("Removing for AtmosphereResource {} cached message {}", uuid, message.getMessage());
+        notifyRemoveCache(broadcasterId, message);
         return this;
     }
 
@@ -240,21 +210,22 @@ private void addMessageIfNotExists(String broadcasterId, String clientId, CacheM
     }
 
     private void addMessage(String broadcasterId, String clientId, CacheMessage message) {
-        ClientQueue clientQueue = messages.get(clientId);
+        ConcurrentLinkedQueue clientQueue = messages.get(clientId);
         if (clientQueue == null) {
-            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;
+            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;
+                }
             }
         }
         notifyAddCache(broadcasterId, message);
-        clientQueue.getQueue().offer(message);
-        clientQueue.getIds().add(message.getId());
+        messages.put(clientId, clientQueue);
     }
 
     private void notifyAddCache(String broadcasterId, CacheMessage message) {
@@ -278,11 +249,11 @@ private void notifyRemoveCache(String broadcasterId, CacheMessage message) {
     }
 
     private boolean hasMessage(String clientId, String messageId) {
-        ClientQueue clientQueue = messages.get(clientId);
-        return clientQueue != null && clientQueue.getIds().contains(messageId);
+        ConcurrentLinkedQueue clientQueue = messages.get(clientId);
+        return clientQueue != null && clientQueue.contains(messageId);
     }
 
-    public Map<String, ClientQueue> messages() {
+    public Map<String, ConcurrentLinkedQueue> messages() {
         return messages;
     }
 
@@ -325,7 +296,7 @@ protected void invalidateExpiredEntries() {
 
         for (String msg : messages().keySet()) {
             if (!activeClients().containsKey(msg)) {
-                messages().remove(msg);
+                messages.remove(msg);
             }
         }
     }

modules/cpr/src/test/java/org/atmosphere/cpr/UUIDBroadcasterCacheTest.java

@@ -76,7 +76,7 @@ public void testBasicCache() throws ExecutionException, InterruptedException, Se
         broadcaster.broadcast("e2").get();
         broadcaster.broadcast("e3").get();
 
-        assertEquals(broadcasterCache.messages().get(ar.uuid()).getQueue().size(), 2);
+        assertEquals(broadcasterCache.messages().get(ar.uuid()).size(), 2);
     }
 
     @Test
@@ -90,7 +90,7 @@ public void addRemoveAddTest() throws ExecutionException, InterruptedException,
         broadcaster.broadcast("e3").get();
 
         assertEquals(broadcasterCache.messages().size(), 1);
-        assertEquals(broadcasterCache.messages().get(ar.uuid()).getQueue().size(), 1);
+        assertEquals(broadcasterCache.messages().get(ar.uuid()).size(), 1);
     }
 
     @Test
@@ -137,7 +137,7 @@ public void run() {
 
         latch.await(10, TimeUnit.SECONDS);
 
-        assertEquals(broadcasterCache.messages().get(ar.uuid()).getQueue().size(), 100);
+        assertEquals(broadcasterCache.messages().get(ar.uuid()).size(), 100);
     }
 
     public final static class AR implements AtmosphereHandler {

modules/native/pom.xml

@@ -11,6 +11,7 @@
     <artifactId>atmosphere-runtime-native</artifactId>
     <packaging>bundle</packaging>
     <version>2.4.10-SNAPSHOT</version>
+
     <name>atmosphere-runtime-native</name>
     <url>https://github.com/Atmosphere/atmosphere</url>
     <build>

pom.xml

@@ -17,8 +17,7 @@
         <connection>scm:git:[email protected]:Atmosphere/atmosphere.git</connection>
         <developerConnection>scm:git:[email protected]:Atmosphere/atmosphere.git</developerConnection>
         <url>http://github.com/Atmosphere/atmosphere</url>
-      <tag>HEAD</tag>
-  </scm>
+   </scm>
     <prerequisites>
         <maven>3.0</maven>
     </prerequisites>