First tentative to for #170 [long-polling] Message losts during concu…

…rrent suspend/broadcast
Atmosphere · Jan 26, 2012 · 76279f3 · 76279f3
1 parent a0c13f0
commit 76279f3
Showing 1 changed file with 11 additions and 0 deletions.
diff --git a/modules/cpr/src/main/java/org/atmosphere/cpr/DefaultBroadcaster.java b/modules/cpr/src/main/java/org/atmosphere/cpr/DefaultBroadcaster.java
@@ -71,6 +71,7 @@
 import static org.atmosphere.cpr.BroadcasterLifeCyclePolicy.ATMOSPHERE_RESOURCE_POLICY.IDLE_DESTROY;
 import static org.atmosphere.cpr.BroadcasterLifeCyclePolicy.ATMOSPHERE_RESOURCE_POLICY.IDLE_RESUME;
 import static org.atmosphere.cpr.BroadcasterLifeCyclePolicy.ATMOSPHERE_RESOURCE_POLICY.NEVER;
+import static org.atmosphere.cpr.HeaderConfig.X_CACHE_DATE;
 
 /**
  * {@link Broadcaster} implementation.
@@ -595,6 +596,7 @@ protected void push(Entry entry) {
                     }
 
                     if (entry.writeLocally) {
+                        pushCachedMessage(r);
                         queueWriteIO(r, finalMsg, entry);
                     }
                 }
@@ -607,6 +609,7 @@ protected void push(Entry entry) {
                 }
 
                 if (entry.writeLocally) {
+                    pushCachedMessage((AtmosphereResource<?, ?>) entry.multipleAtmoResources);
                     queueWriteIO((AtmosphereResource<?, ?>) entry.multipleAtmoResources, finalMsg, entry);
                 }
             } else if (entry.multipleAtmoResources instanceof Set) {
@@ -620,6 +623,7 @@ protected void push(Entry entry) {
                     }
 
                     if (entry.writeLocally) {
+                        pushCachedMessage(r);
                         queueWriteIO(r, finalMsg, entry);
                     }
                 }
@@ -630,6 +634,13 @@ protected void push(Entry entry) {
         }
     }
 
+    protected void pushCachedMessage(AtmosphereResource<?,?> r) {
+        AtmosphereResourceImpl ari = AtmosphereResourceImpl.class.cast(r);
+        // Check lost message send from the last 2 seconds.
+        ari.getRequest(false).setAttribute(X_CACHE_DATE, System.currentTimeMillis() - 2000);
+        retrieveTrackedBroadcast(r, new AtmosphereResourceEventImpl(ari, false, false));
+    }
+
     protected void queueWriteIO(AtmosphereResource<?, ?> r, Object finalMsg, Entry entry) throws InterruptedException {
         asyncWriteQueue.put(new AsyncWriteToken(r, finalMsg, entry.future, entry.originalMessage));
     }