[jbosscache-commits] JBoss Cache SVN: r4604 - core/trunk/src/main/java/org/jboss/cache/interceptors.

[jbosscache-commits at lists.jboss.org]

Author: manik.surtani at jboss.com
Date: 2007-10-12 10:24:34 -0400 (Fri, 12 Oct 2007)
New Revision: 4604

Modified:
   core/trunk/src/main/java/org/jboss/cache/interceptors/DataGravitatorInterceptor.java
Log:
JBCACHE-1186 - Make use of JGroups 2.6 RspFilter to filter responses that come back

Modified: core/trunk/src/main/java/org/jboss/cache/interceptors/DataGravitatorInterceptor.java
===================================================================
--- core/trunk/src/main/java/org/jboss/cache/interceptors/DataGravitatorInterceptor.java	2007-10-12 14:23:56 UTC (rev 4603)
+++ core/trunk/src/main/java/org/jboss/cache/interceptors/DataGravitatorInterceptor.java	2007-10-12 14:24:34 UTC (rev 4604)
@@ -17,6 +17,7 @@
 import org.jboss.cache.buddyreplication.BuddyManager;
 import org.jboss.cache.buddyreplication.GravitateResult;
 import org.jboss.cache.config.Configuration;
+import org.jboss.cache.loader.ClusteredCacheLoader;
 import org.jboss.cache.marshall.MethodCall;
 import org.jboss.cache.marshall.MethodCallFactory;
 import org.jboss.cache.marshall.MethodDeclarations;
@@ -25,6 +26,7 @@
 import org.jboss.cache.transaction.TransactionEntry;
 import org.jgroups.Address;
 import org.jgroups.blocks.GroupRequest;
+import org.jgroups.blocks.RspFilter;
 
 import java.util.ArrayList;
 import java.util.Collection;
@@ -321,7 +323,7 @@
       MethodCall dGrav = MethodCallFactory.create(MethodDeclarations.dataGravitationMethod, fqn, searchSubtrees);
       // doing a GET_ALL is crappy but necessary since JGroups' GET_FIRST could return null results from nodes that do
       // not have either the primary OR backup, and stop polling other valid nodes.
-      List resps = cache.getRPCManager().callRemoteMethods(mbrs, dGrav, GroupRequest.GET_ALL, true, buddyManager.getBuddyCommunicationTimeout());
+      List resps = cache.getRPCManager().callRemoteMethods(mbrs, dGrav, GroupRequest.GET_FIRST, true, buddyManager.getBuddyCommunicationTimeout(), new ResponseValidityFilter());
       if (log.isTraceEnabled())
       {
          log.trace("got responses " + resps);
@@ -382,7 +384,7 @@
       }
    }
 
-   private void createNodesLocally(Fqn fqn, Map data) throws CacheException
+   private void createNodesLocally(Fqn<Object> fqn, Map data) throws CacheException
    {
       int treeNodeSize;
       if ((treeNodeSize = fqn.size()) == 0) return;
@@ -390,7 +392,7 @@
       for (int i = 0; i < treeNodeSize; i++)
       {
          Object child_name = fqn.get(i);
-         NodeSPI child_node = n.addChildDirect(new Fqn(child_name));
+         NodeSPI child_node = n.addChildDirect(new Fqn<Object>(child_name));
          if (child_node == null)
          {
             if (log.isTraceEnabled())
@@ -418,6 +420,7 @@
       return (Fqn) args[MethodDeclarations.isCrudMethod(methodId) ? 1 : 0];
    }
 
+   @SuppressWarnings("unchecked")
    private boolean localBackupExists(Fqn fqn)
    {
       boolean exists = false;
@@ -459,6 +462,7 @@
       return data;
    }
 
+   @SuppressWarnings("unchecked")
    private Collection<Node> getBackupRootCollection()
    {
       NodeSPI backupRoot = cache.peek(BuddyManager.BUDDY_BACKUP_SUBTREE_FQN, true);
@@ -479,5 +483,23 @@
       }
    }
 
+   public static class ResponseValidityFilter implements RspFilter
+   {
+      private int numValidResponses = 0;
 
+      public boolean isAcceptable(Object object, Address address)
+      {
+         if (!(object instanceof GravitateResult)) return false;
+
+         GravitateResult response = (GravitateResult) object;
+         if (response.isDataFound()) numValidResponses++;
+
+         return response.isDataFound();
+      }
+
+      public boolean needMoreResponses()
+      {
+         return numValidResponses < 1;
+      }
+   }
 }