[jboss-cvs] JBossAS SVN: r72957 - trunk/cluster/src/main/org/jboss/ha/framework/server.

[jboss-cvs-commits at lists.jboss.org]

Author: bstansberry at jboss.com
Date: 2008-05-01 16:49:07 -0400 (Thu, 01 May 2008)
New Revision: 72957

Modified:
   trunk/cluster/src/main/org/jboss/ha/framework/server/DistributedStateImpl.java
Log:
Support asynchronous calls

Modified: trunk/cluster/src/main/org/jboss/ha/framework/server/DistributedStateImpl.java
===================================================================
--- trunk/cluster/src/main/org/jboss/ha/framework/server/DistributedStateImpl.java	2008-05-01 20:25:35 UTC (rev 72956)
+++ trunk/cluster/src/main/org/jboss/ha/framework/server/DistributedStateImpl.java	2008-05-01 20:49:07 UTC (rev 72957)
@@ -36,6 +36,7 @@
 import org.jboss.cache.CacheException;
 import org.jboss.cache.Fqn;
 import org.jboss.cache.Node;
+import org.jboss.cache.config.Configuration.CacheMode;
 import org.jboss.cache.notifications.annotation.CacheListener;
 import org.jboss.cache.notifications.annotation.NodeModified;
 import org.jboss.cache.notifications.event.NodeModifiedEvent;
@@ -71,6 +72,7 @@
 //   protected MBeanServer mbeanServer = null;
    protected String name = null;
    protected Cache cache;
+   protected boolean replAsync;
 
    public static final String ROOT = "__DISTRIBUTED_STATE__";
 
@@ -130,6 +132,8 @@
 
    public void startService() throws Exception
    {
+      if (cache == null)
+         throw new IllegalStateException("No clustered cache available");
 //      super.startService();
    }
 
@@ -214,11 +218,27 @@
       return cache;
    }
 
+   /**
+    * Sets the cache to use.
+    * 
+    * @param cache the cache
+    * 
+    * @throws IllegalStateException if the cache isn't configured for replication
+    */
    public void setClusteredCache(Cache cache)
    {
       this.cache = cache;
-      if (this.cache != null)
+      if (this.cache != null) 
+      {
       	this.cache.addCacheListener(this);
+      	CacheMode cm = cache.getConfiguration().getCacheMode();
+      	if (CacheMode.REPL_ASYNC == cm)
+      	   this.replAsync = true;
+      	else if (CacheMode.REPL_SYNC == cm)
+           this.replAsync = true;
+      	else
+      	   throw new IllegalStateException("Cache must be configured for replication, not " + cm);
+      }
    }
 
    // DistributedState implementation ----------------------------------------------
@@ -232,7 +252,7 @@
    public void set(String category, Serializable key, Serializable value)
          throws Exception
    {
-      cache.put(buildFqn(category), key, value);
+      set(category, key, value, true);
    }
 
    /*
@@ -246,8 +266,14 @@
    public void set(String category, Serializable key, Serializable value,
          boolean asynchronousCall) throws Exception
    {
-      // TODO does not support asynchronousCall yet. TreeCache cannot switch this on the fly
-      set(category, key, value);
+      if (replAsync != asynchronousCall)
+      {
+         if (asynchronousCall)
+            cache.getInvocationContext().getOptionOverrides().setForceAsynchronous(true);
+         else
+            cache.getInvocationContext().getOptionOverrides().setForceSynchronous(true);
+      }
+      cache.put(buildFqn(category), key, value);      
    }
 
    /*
@@ -267,10 +293,7 @@
     * (non-Javadoc)
     *
     * @see org.jboss.ha.framework.interfaces.DistributedState#remove(java.lang.String,
-    *      java.io.Serializable, boolean) @param asynchronousCall is not
-    *      supported yet. TreeCache cannot switch this on the fly. Will take
-    *      value from TreeCache-config instead. @return returns null in case of
-    *      CacheException!
+    *      java.io.Serializable, boolean)
     */
    public Serializable remove(String category, Serializable key,
          boolean asynchronousCall) throws Exception
@@ -278,6 +301,13 @@
       Serializable retVal = get(category, key);
       if(retVal != null)
       {
+         if (replAsync != asynchronousCall)
+         {
+            if (asynchronousCall)
+               cache.getInvocationContext().getOptionOverrides().setForceAsynchronous(true);
+            else
+               cache.getInvocationContext().getOptionOverrides().setForceSynchronous(true);
+         }
          cache.remove(buildFqn(category), key);
       }
 	  return retVal;