[jboss-cvs] JBossAS SVN: r105394 - projects/cluster/ha-server-core/trunk/src/main/java/org/jboss/ha/core/framework/server.

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

Author: bstansberry at jboss.com
Date: 2010-05-29 09:40:51 -0400 (Sat, 29 May 2010)
New Revision: 105394

Modified:
   projects/cluster/ha-server-core/trunk/src/main/java/org/jboss/ha/core/framework/server/CoreGroupCommunicationService.java
Log:
Properly use stateIdPrefix

Modified: projects/cluster/ha-server-core/trunk/src/main/java/org/jboss/ha/core/framework/server/CoreGroupCommunicationService.java
===================================================================
--- projects/cluster/ha-server-core/trunk/src/main/java/org/jboss/ha/core/framework/server/CoreGroupCommunicationService.java	2010-05-29 02:43:08 UTC (rev 105393)
+++ projects/cluster/ha-server-core/trunk/src/main/java/org/jboss/ha/core/framework/server/CoreGroupCommunicationService.java	2010-05-29 13:40:51 UTC (rev 105394)
@@ -893,7 +893,7 @@
       this.dispatcher.setMembershipListener(new MembershipListenerImpl());
       if (this.stateIdPrefix != null)
       {
-         this.dispatcher.setMessageListener(new MessageListenerAdapter());
+         this.dispatcher.setMessageListener(new MessageListenerImpl());
       }
       
       if (!this.channel.isConnected())
@@ -1933,7 +1933,7 @@
    /**
     * Handles MessageListener callbacks from the JGroups layer.
     */
-   private class MessageListenerAdapter
+   private class MessageListenerImpl
          implements ExtendedMessageListener
    {
 
@@ -1944,10 +1944,11 @@
       
       public void getState(String state_id, OutputStream ostream)
       {
-         // FIXME refactor to share logic
-         CoreGroupCommunicationService.this.log.debug("getState called for service " + state_id);
+         String serviceName = extractServiceName(state_id);
          
-         StateTransferProvider provider = stateProviders.get(state_id);
+         CoreGroupCommunicationService.this.log.debug("getState called for service " + serviceName);
+         
+         StateTransferProvider provider = stateProviders.get(serviceName);
          if (provider != null)
          {
             MarshalledValueOutputStream mvos = null;
@@ -1960,7 +1961,7 @@
             }
             catch (Exception ex)
             {
-               CoreGroupCommunicationService.this.log.error("getState failed for service " + state_id, ex);
+               CoreGroupCommunicationService.this.log.error("getState failed for service " + serviceName, ex);
             }
             finally
             {
@@ -1982,9 +1983,11 @@
 
       public byte[] getState(String state_id)
       {
-         CoreGroupCommunicationService.this.log.debug("getState called for service " + state_id);
+         String serviceName = extractServiceName(state_id);
          
-         StateTransferProvider provider = stateProviders.get(state_id);
+         CoreGroupCommunicationService.this.log.debug("getState called for service " + serviceName);
+         
+         StateTransferProvider provider = stateProviders.get(serviceName);
          if (provider != null)
          {
             MarshalledValueOutputStream mvos = null;
@@ -2000,7 +2003,7 @@
             }
             catch (Exception ex)
             {
-               CoreGroupCommunicationService.this.log.error("getState failed for service " + state_id, ex);
+               CoreGroupCommunicationService.this.log.error("getState failed for service " + serviceName, ex);
             }
             finally
             {
@@ -2023,11 +2026,13 @@
 
       public void setState(String state_id, byte[] state)
       {
-         StateTransferTask task = CoreGroupCommunicationService.this.stateTransferTasks.get(state_id);
+         String serviceName = extractServiceName(state_id);
+         
+         StateTransferTask task = CoreGroupCommunicationService.this.stateTransferTasks.get(serviceName);
          if (task == null)
          {
             CoreGroupCommunicationService.this.log.warn("No " + StateTransferTask.class.getSimpleName() + 
-                  " registered to receive state for service " + state_id);
+                  " registered to receive state for service " + serviceName);
          }
          else
          {
@@ -2037,11 +2042,13 @@
 
       public void setState(String state_id, InputStream istream)
       {
-         StateTransferTask task = CoreGroupCommunicationService.this.stateTransferTasks.get(state_id);
+         String serviceName = extractServiceName(state_id);
+         
+         StateTransferTask task = CoreGroupCommunicationService.this.stateTransferTasks.get(serviceName);
          if (task == null)
          {
             CoreGroupCommunicationService.this.log.warn("No " + StateTransferTask.class.getSimpleName() + 
-                  " registered to receive state for service " + state_id);
+                  " registered to receive state for service " + serviceName);
          }
          else
          {
@@ -2068,6 +2075,16 @@
       {
          throw new UnsupportedOperationException("Only partial state transfer (with a state_id) is supported");
       }
+
+      private String extractServiceName(String state_id)
+      {
+         if (!state_id.startsWith(CoreGroupCommunicationService.this.stateIdPrefix))
+         {
+            throw new IllegalArgumentException("Unknown state_id " + state_id + 
+                  " -- must start with " + CoreGroupCommunicationService.this.stateIdPrefix);
+         }
+         return state_id.substring(CoreGroupCommunicationService.this.stateIdPrefix.length());
+      }
       
    }
    
@@ -2103,7 +2120,8 @@
             long start, stop;
             this.isStateSet = false;
             start = System.currentTimeMillis();
-            boolean rc = CoreGroupCommunicationService.this.getChannel().getState(null, serviceName, CoreGroupCommunicationService.this.getStateTransferTimeout());
+            String state_id = CoreGroupCommunicationService.this.stateIdPrefix + serviceName;
+            boolean rc = CoreGroupCommunicationService.this.getChannel().getState(null, state_id, CoreGroupCommunicationService.this.getStateTransferTimeout());
             if (rc)
             {
                synchronized (this)