[jboss-cvs] JBoss Messaging SVN: r1543 - in branches/Branch_Client_Failover_Experiment/src/main/org/jboss: jms/server/endpoint messaging/core/plugin/contract messaging/core/plugin/postoffice messaging/core/plugin/postoffice/cluster

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

Author: clebert.suconic at jboss.com
Date: 2006-11-01 18:34:39 -0500 (Wed, 01 Nov 2006)
New Revision: 1543

Modified:
   branches/Branch_Client_Failover_Experiment/src/main/org/jboss/jms/server/endpoint/ServerSessionEndpoint.java
   branches/Branch_Client_Failover_Experiment/src/main/org/jboss/messaging/core/plugin/contract/PostOffice.java
   branches/Branch_Client_Failover_Experiment/src/main/org/jboss/messaging/core/plugin/postoffice/DefaultPostOffice.java
   branches/Branch_Client_Failover_Experiment/src/main/org/jboss/messaging/core/plugin/postoffice/cluster/DefaultClusteredPostOffice.java
Log:
http://jira.jboss.org/jira/browse/JBMESSAGING-519 - Changing Queue from one node to another instead of keeping node on the same queue.

Modified: branches/Branch_Client_Failover_Experiment/src/main/org/jboss/jms/server/endpoint/ServerSessionEndpoint.java
===================================================================
--- branches/Branch_Client_Failover_Experiment/src/main/org/jboss/jms/server/endpoint/ServerSessionEndpoint.java	2006-11-01 21:08:42 UTC (rev 1542)
+++ branches/Branch_Client_Failover_Experiment/src/main/org/jboss/jms/server/endpoint/ServerSessionEndpoint.java	2006-11-01 23:34:39 UTC (rev 1543)
@@ -202,7 +202,27 @@
          {
             selector = new Selector(selectorString);
          }
-   
+
+
+         // fail over channel
+         if (oldchannelID>=0)
+         {
+             PostOffice postOfficeToUse = null;
+             if (jmsDestination.isTopic())
+             {
+                 postOfficeToUse = topicPostOffice;
+             }
+             else
+             {
+                 postOfficeToUse = queuePostOffice;
+             }
+             binding = postOfficeToUse.getBindingforChannelId(-1,oldchannelID);
+             if (binding==null)
+             {
+                 throw new JMSException("Can't find failed over channel " + oldchannelID);
+             }
+         }
+         else
          if (jmsDestination.isTopic())
          {
             if (subscriptionName == null)

Modified: branches/Branch_Client_Failover_Experiment/src/main/org/jboss/messaging/core/plugin/contract/PostOffice.java
===================================================================
--- branches/Branch_Client_Failover_Experiment/src/main/org/jboss/messaging/core/plugin/contract/PostOffice.java	2006-11-01 21:08:42 UTC (rev 1542)
+++ branches/Branch_Client_Failover_Experiment/src/main/org/jboss/messaging/core/plugin/contract/PostOffice.java	2006-11-01 23:34:39 UTC (rev 1543)
@@ -73,6 +73,8 @@
    Binding getBindingForQueueName(int nodeID, String queueName) throws Exception;
    Binding getBindingForQueueName(String queueName) throws Exception;
 
+   Binding getBindingforChannelId(int parameterNodeId, long channelId) throws Exception;    
+
    /**
     * Route a reference.
     * @param ref

Modified: branches/Branch_Client_Failover_Experiment/src/main/org/jboss/messaging/core/plugin/postoffice/DefaultPostOffice.java
===================================================================
--- branches/Branch_Client_Failover_Experiment/src/main/org/jboss/messaging/core/plugin/postoffice/DefaultPostOffice.java	2006-11-01 21:08:42 UTC (rev 1542)
+++ branches/Branch_Client_Failover_Experiment/src/main/org/jboss/messaging/core/plugin/postoffice/DefaultPostOffice.java	2006-11-01 23:34:39 UTC (rev 1543)
@@ -288,7 +288,6 @@
 
 
          log.info("Returned " + binding);
-         
          return binding;
       }
       finally
@@ -296,6 +295,47 @@
          lock.readLock().release();
       }
    }
+
+   public Binding getBindingforChannelId(int parameterNodeId, long channelId) throws Exception
+   {
+       log.info("DefaultPostOffice::getBindingforChannelId(" + parameterNodeId + " ," + channelId + ")");
+       if (parameterNodeId<0) parameterNodeId=this.nodeId;
+
+       lock.readLock().acquire();
+
+       try
+       {
+          Map nameMap = (Map)nameMaps.get(new Integer(parameterNodeId));
+
+          Binding binding = null;
+
+          if (nameMap != null)
+          {
+             for (Iterator iterbindings = nameMap.values().iterator();iterbindings.hasNext();)
+             {
+                 Binding itemBinding = (Binding)iterbindings.next();
+                 if (itemBinding.getQueue().getChannelID()==channelId)
+                 {
+                     binding=itemBinding;
+                     break;
+                 }
+             }
+
+          }
+          else
+          {
+              log.info("nameMap is null");
+          }
+
+
+          log.info("Returned " + binding);
+          return binding;
+       }
+       finally
+       {
+          lock.readLock().release();
+       }
+   }
    
    public void recover() throws Exception
    {

Modified: branches/Branch_Client_Failover_Experiment/src/main/org/jboss/messaging/core/plugin/postoffice/cluster/DefaultClusteredPostOffice.java
===================================================================
--- branches/Branch_Client_Failover_Experiment/src/main/org/jboss/messaging/core/plugin/postoffice/cluster/DefaultClusteredPostOffice.java	2006-11-01 21:08:42 UTC (rev 1542)
+++ branches/Branch_Client_Failover_Experiment/src/main/org/jboss/messaging/core/plugin/postoffice/cluster/DefaultClusteredPostOffice.java	2006-11-01 23:34:39 UTC (rev 1543)
@@ -1129,12 +1129,20 @@
            RemoteQueueStub stub = (RemoteQueueStub)binding.getQueue();
            this.removeBinding(nodeId,(String)entry.getKey());
 
-           Binding newBinding = this.createBinding(this.nodeId,nodeId,binding.getCondition(),stub.getName(),stub.getChannelID(),stub.getFilter(),stub.isRecoverable());
+           this.deleteBinding(nodeId,(String)entry.getKey());
+
+           // todo: remove Binding from cluster
+
+           Binding newBinding = this.createBinding(this.nodeId,this.nodeId,binding.getCondition(),stub.getName(),stub.getChannelID(),stub.getFilter(),stub.isRecoverable());
+
+           insertBinding(newBinding);
+
            LocalClusteredQueue clusteredQueue = (LocalClusteredQueue )newBinding.getQueue();
            clusteredQueue.deactivate();
            clusteredQueue.load();
            clusteredQueue.activate();
            addBinding(newBinding);
+           // todo: send new Binding into Cluster
        }
    }