4:42 a.m.
Author: nfilotto
Date: 2010-10-11 05:42:23 -0400 (Mon, 11 Oct 2010)
New Revision: 3283
Added:
kernel/trunk/exo.kernel.component.common/src/main/java/org/exoplatform/services/rpc/TopologyChangeEvent.java
kernel/trunk/exo.kernel.component.common/src/main/java/org/exoplatform/services/rpc/TopologyChangeListener.java
Modified:
kernel/trunk/exo.kernel.component.common/src/main/java/org/exoplatform/services/rpc/RPCService.java
kernel/trunk/exo.kernel.component.common/src/main/java/org/exoplatform/services/rpc/impl/RPCServiceImpl.java
kernel/trunk/exo.kernel.component.common/src/test/java/org/exoplatform/services/rpc/impl/TestRPCServiceImpl.java
Log:
EXOJCR-967: We now allow to listen to any topology changes
Modified:
kernel/trunk/exo.kernel.component.common/src/main/java/org/exoplatform/services/rpc/RPCService.java
===================================================================
---
kernel/trunk/exo.kernel.component.common/src/main/java/org/exoplatform/services/rpc/RPCService.java 2010-10-08
21:00:13 UTC (rev 3282)
+++
kernel/trunk/exo.kernel.component.common/src/main/java/org/exoplatform/services/rpc/RPCService.java 2010-10-11
09:42:23 UTC (rev 3283)
@@ -130,5 +130,21 @@
* @return <code>true</code> if the coordinator is the coordinator,
<code>false</code> otherwise
* throws RPCException in case the {@link RPCService} is in an illegal state
*/
- boolean isCoordinator() throws RPCException;
+ boolean isCoordinator() throws RPCException;
+
+ /**
+ * Register a new {@link TopologyChangeListener}
+ * @param listener the listener to be registered
+ * @throws SecurityException if the {@link SecurityManager} is installed and the call
method
+ * doesn't have the {@link RuntimePermission}
<code>ACCESS_RPC_SERVICE_PERMISSION</code>
+ */
+ void registerTopologyChangeListener(TopologyChangeListener listener) throws
SecurityException;
+
+ /**
+ * Unregister a {@link TopologyChangeListener} if it exists
+ * @param listener the listener to unregister
+ * @throws SecurityException if the {@link SecurityManager} is installed and the call
method
+ * doesn't have the {@link RuntimePermission}
<code>ACCESS_RPC_SERVICE_PERMISSION</code>
+ */
+ void unregisterTopologyChangeListener(TopologyChangeListener listener) throws
SecurityException;
}
\ No newline at end of file
Added:
kernel/trunk/exo.kernel.component.common/src/main/java/org/exoplatform/services/rpc/TopologyChangeEvent.java
===================================================================
---
kernel/trunk/exo.kernel.component.common/src/main/java/org/exoplatform/services/rpc/TopologyChangeEvent.java
(rev 0)
+++
kernel/trunk/exo.kernel.component.common/src/main/java/org/exoplatform/services/rpc/TopologyChangeEvent.java 2010-10-11
09:42:23 UTC (rev 3283)
@@ -0,0 +1,70 @@
+/*
+ * Copyright (C) 2010 eXo Platform SAS.
+ *
+ * This is free software; you can redistribute it and/or modify it
+ * under the terms of the GNU Lesser General Public License as
+ * published by the Free Software Foundation; either version 2.1 of
+ * the License, or (at your option) any later version.
+ *
+ * This software is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public
+ * License along with this software; if not, write to the Free
+ * Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA
+ * 02110-1301 USA, or see the FSF site: http://www.fsf.org.
+ */
+package org.exoplatform.services.rpc;
+
+/**
+ * The event triggered anytime the cluster has changed of topology
+ * @author <a href="mailto:nicolas.filotto@exoplatform.com">Nicolas
Filotto</a>
+ * @version $Id$
+ *
+ */
+public class TopologyChangeEvent
+{
+
+ /**
+ * Indicates whether the current node is the coordinator after this topology change
+ */
+ private final boolean coordinator;
+
+ /**
+ * Indicates whether the coordinator has changed
+ */
+ private final boolean coordinatorHasChanged;
+
+ /**
+ * Default constructor
+ * @param coordinatorHasChanged this parameter is set to <code>true</code>
if the
+ * coordinator has changed, <code>false</code> otherwise
+ * @param coordinator this parameter is set to <code>true</code> if the
current node
+ * is the coordinator, <code>false</code> otherwise
+ */
+ public TopologyChangeEvent(boolean coordinatorHasChanged, boolean coordinator)
+ {
+ this.coordinator = coordinator;
+ this.coordinatorHasChanged = coordinatorHasChanged;
+ }
+
+ /**
+ * Indicates whether the current node is the coordinator or not after the topology
change
+ * @return <code>true</code> if the current node is the coordinator,
<code>false</code> otherwise.
+ */
+ public boolean isCoordinator()
+ {
+ return coordinator;
+ }
+
+ /**
+ * Indicates whether the coordinator has changed after the topology change
+ * @return <code>true</code> if the coordinator has changed,
<code>false</code> otherwise.
+ */
+ public boolean isCoordinatorHasChanged()
+ {
+ return coordinatorHasChanged;
+ }
+}
Added:
kernel/trunk/exo.kernel.component.common/src/main/java/org/exoplatform/services/rpc/TopologyChangeListener.java
===================================================================
---
kernel/trunk/exo.kernel.component.common/src/main/java/org/exoplatform/services/rpc/TopologyChangeListener.java
(rev 0)
+++
kernel/trunk/exo.kernel.component.common/src/main/java/org/exoplatform/services/rpc/TopologyChangeListener.java 2010-10-11
09:42:23 UTC (rev 3283)
@@ -0,0 +1,38 @@
+/*
+ * Copyright (C) 2010 eXo Platform SAS.
+ *
+ * This is free software; you can redistribute it and/or modify it
+ * under the terms of the GNU Lesser General Public License as
+ * published by the Free Software Foundation; either version 2.1 of
+ * the License, or (at your option) any later version.
+ *
+ * This software is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public
+ * License along with this software; if not, write to the Free
+ * Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA
+ * 02110-1301 USA, or see the FSF site: http://www.fsf.org.
+ */
+package org.exoplatform.services.rpc;
+
+/**
+ * This interface is used to allow some components to be notified anytime the topology
+ * of the cluster changes, in other words any time a member of the cluster leave or
join
+ * the cluster.
+ *
+ * @author <a href="mailto:nicolas.filotto@exoplatform.com">Nicolas
Filotto</a>
+ * @version $Id$
+ *
+ */
+public interface TopologyChangeListener
+{
+
+ /**
+ * Called anytime the topology of the cluster changes
+ * @param event the event triggered when the topology changes
+ */
+ void onChange(TopologyChangeEvent event);
+}
Modified:
kernel/trunk/exo.kernel.component.common/src/main/java/org/exoplatform/services/rpc/impl/RPCServiceImpl.java
===================================================================
---
kernel/trunk/exo.kernel.component.common/src/main/java/org/exoplatform/services/rpc/impl/RPCServiceImpl.java 2010-10-08
21:00:13 UTC (rev 3282)
+++
kernel/trunk/exo.kernel.component.common/src/main/java/org/exoplatform/services/rpc/impl/RPCServiceImpl.java 2010-10-11
09:42:23 UTC (rev 3283)
@@ -29,6 +29,8 @@
import org.exoplatform.services.rpc.RPCException;
import org.exoplatform.services.rpc.RPCService;
import org.exoplatform.services.rpc.RemoteCommand;
+import org.exoplatform.services.rpc.TopologyChangeEvent;
+import org.exoplatform.services.rpc.TopologyChangeListener;
import org.jgroups.Address;
import org.jgroups.Channel;
import org.jgroups.ChannelException;
@@ -61,6 +63,7 @@
import java.util.List;
import java.util.Map;
import java.util.Vector;
+import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.CountDownLatch;
/**
@@ -103,7 +106,7 @@
* The default value of the cluster name
*/
public static final String CLUSTER_NAME = "RPCService-Cluster";
-
+
/**
* The configurator used to create the JGroups Channel
*/
@@ -151,6 +154,11 @@
* incoming messages.
*/
private final CountDownLatch startSignal = new CountDownLatch(1);
+
+ /**
+ * All the registered {@link TopologyChangeListener}
+ */
+ private final List<TopologyChangeListener> listeners = new
CopyOnWriteArrayList<TopologyChangeListener>();
/**
* Current State of the {@link RPCServiceImpl}
@@ -434,11 +442,35 @@
public void viewAccepted(View view)
{
this.members = view.getMembers();
+ Address currentCoordinator = coordinator;
this.coordinator = members != null && members.size() > 0 ?
members.get(0) : null;
this.isCoordinator = coordinator != null &&
coordinator.equals(channel.getLocalAddress());
+ onTopologyChange(currentCoordinator != null &&
!currentCoordinator.equals(coordinator));
}
/**
+ * Called anytime the topology has changed, this method will notify all the listeners
+ * currently registered
+ * @param coordinatorHasChanged this parameter is set to <code>true</code>
if the
+ * coordinator has changed, <code>false</code> otherwise
+ */
+ private void onTopologyChange(boolean coordinatorHasChanged)
+ {
+ TopologyChangeEvent event = new TopologyChangeEvent(coordinatorHasChanged,
isCoordinator);
+ for (TopologyChangeListener listener : listeners)
+ {
+ try
+ {
+ listener.onChange(event);
+ }
+ catch (Exception e)
+ {
+ LOG.warn("An error occurs with the listener of type " +
listener.getClass(), e);
+ }
+ }
+ }
+
+ /**
* {@inheritDoc}
*/
public synchronized RemoteCommand registerCommand(RemoteCommand command)
@@ -517,6 +549,40 @@
}
/**
+ * {@inheritDoc}
+ */
+ public void registerTopologyChangeListener(TopologyChangeListener listener) throws
SecurityException
+ {
+ SecurityManager security = System.getSecurityManager();
+ if (security != null)
+ {
+ security.checkPermission(RPCService.ACCESS_RPC_SERVICE_PERMISSION);
+ }
+ if (listener == null)
+ {
+ return;
+ }
+ listeners.add(listener);
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ public void unregisterTopologyChangeListener(TopologyChangeListener listener) throws
SecurityException
+ {
+ SecurityManager security = System.getSecurityManager();
+ if (security != null)
+ {
+ security.checkPermission(RPCService.ACCESS_RPC_SERVICE_PERMISSION);
+ }
+ if (listener == null)
+ {
+ return;
+ }
+ listeners.remove(listener);
+ }
+
+ /**
* Gives the {@link RemoteCommand} corresponding to the given id
* @param commandId the command id of the command to retrieve
* @return the corresponding {@link RemoteCommand}
Modified:
kernel/trunk/exo.kernel.component.common/src/test/java/org/exoplatform/services/rpc/impl/TestRPCServiceImpl.java
===================================================================
---
kernel/trunk/exo.kernel.component.common/src/test/java/org/exoplatform/services/rpc/impl/TestRPCServiceImpl.java 2010-10-08
21:00:13 UTC (rev 3282)
+++
kernel/trunk/exo.kernel.component.common/src/test/java/org/exoplatform/services/rpc/impl/TestRPCServiceImpl.java 2010-10-11
09:42:23 UTC (rev 3283)
@@ -25,6 +25,8 @@
import org.exoplatform.services.rpc.RPCException;
import org.exoplatform.services.rpc.RemoteCommand;
import org.exoplatform.services.rpc.SingleMethodCallCommand;
+import org.exoplatform.services.rpc.TopologyChangeEvent;
+import org.exoplatform.services.rpc.TopologyChangeListener;
import org.exoplatform.services.rpc.impl.RPCServiceImpl.MemberHasLeftException;
import org.exoplatform.test.BasicTestCase;
import org.jgroups.Address;
@@ -519,7 +521,7 @@
paramConf.setName(RPCServiceImpl.PARAM_JGROUPS_CONFIG);
paramConf.setValue("jar:/conf/portal/udp.xml");
params.addParameter(paramConf);
- RPCServiceImpl service1 = null, service2 = null;
+ RPCServiceImpl service1 = null, service2 = null;
try
{
service1 = new RPCServiceImpl(container.getContext(), params, configManager);
@@ -591,7 +593,7 @@
public String execute(Serializable[] args) throws Throwable
{
- Thread.sleep(30000);
+ Thread.sleep(3000);
return "OldCoordinator";
}
});
@@ -675,9 +677,31 @@
return "NewCoordinator";
}
};
- service2.registerCommand(LongTask);
+ service2.registerCommand(LongTask);
+ MyListener listener1 = new MyListener();
+ service1.registerTopologyChangeListener(listener1);
+ MyListener listener2 = new MyListener();
+ service2.registerTopologyChangeListener(listener2);
+ assertFalse(listener1.coordinatorHasChanged);
+ assertFalse(listener1.isCoordinator);
+ assertEquals(0, listener1.count);
+ assertFalse(listener2.coordinatorHasChanged);
+ assertFalse(listener2.isCoordinator);
+ assertEquals(0, listener2.count);
service1.start();
+ assertFalse(listener1.coordinatorHasChanged);
+ assertTrue(listener1.isCoordinator);
+ assertEquals(1, listener1.count);
+ assertFalse(listener2.coordinatorHasChanged);
+ assertFalse(listener2.isCoordinator);
+ assertEquals(0, listener2.count);
service2.start();
+ assertFalse(listener1.coordinatorHasChanged);
+ assertTrue(listener1.isCoordinator);
+ assertEquals(2, listener1.count);
+ assertFalse(listener2.coordinatorHasChanged);
+ assertFalse(listener2.isCoordinator);
+ assertEquals(1, listener2.count);
assertEquals(true, service1.isCoordinator());
assertEquals(false, service2.isCoordinator());
List<Object> result;
@@ -740,6 +764,12 @@
};
t.start();
service1.stop();
+ assertFalse(listener1.coordinatorHasChanged);
+ assertTrue(listener1.isCoordinator);
+ assertEquals(2, listener1.count);
+ assertTrue(listener2.coordinatorHasChanged);
+ assertTrue(listener2.isCoordinator);
+ assertEquals(2, listener2.count);
doneSignal.await();
assertNull(error.get() != null ? error.get().getMessage() : "",
error.get());
result = service2.excecuteCommand(allMembers, OK, true, 0);
@@ -1081,4 +1111,22 @@
return (short)(values[0] + 3);
}
}
+
+ private static class MyListener implements TopologyChangeListener
+ {
+
+ private boolean coordinatorHasChanged;
+ private boolean isCoordinator;
+ private int count;
+
+ /**
+ * @see
org.exoplatform.services.rpc.TopologyChangeListener#onChange(org.exoplatform.services.rpc.TopologyChangeEvent)
+ */
+ public void onChange(TopologyChangeEvent event)
+ {
+ this.coordinatorHasChanged = event.isCoordinatorHasChanged();
+ this.isCoordinator = event.isCoordinator();
+ count++;
+ }
+ }
}