Author: rfeng
Date: Mon Oct 12 19:31:48 2009
New Revision: 824468
URL: http://svn.apache.org/viewvc?rev=824468&view=rev
Log:
Make sure that entries are replicated to all nodes (not just at least a backup
node)
Added:
tuscany/java/sca/modules/endpoint-tribes/src/test/java/org/apache/tuscany/sca/endpoint/tribes/ReplicatedEndpointRegistryTestCase.java
- copied, changed from r824393,
tuscany/java/sca/modules/endpoint-tribes/src/test/java/org/apache/tuscany/sca/endpoint/tribes/Test.java
Removed:
tuscany/java/sca/modules/endpoint-tribes/src/test/java/org/apache/tuscany/sca/endpoint/tribes/Test.java
Modified:
tuscany/java/sca/modules/endpoint-tribes/src/main/java/org/apache/tuscany/sca/endpoint/tribes/AbstractReplicatedMap.java
tuscany/java/sca/modules/endpoint-tribes/src/main/java/org/apache/tuscany/sca/endpoint/tribes/ReplicatedEndpointRegistry.java
tuscany/java/sca/modules/endpoint-tribes/src/main/java/org/apache/tuscany/sca/endpoint/tribes/ReplicatedMap.java
Modified:
tuscany/java/sca/modules/endpoint-tribes/src/main/java/org/apache/tuscany/sca/endpoint/tribes/AbstractReplicatedMap.java
URL:
http://svn.apache.org/viewvc/tuscany/java/sca/modules/endpoint-tribes/src/main/java/org/apache/tuscany/sca/endpoint/tribes/AbstractReplicatedMap.java?rev=824468&r1=824467&r2=824468&view=diff
==============================================================================
---
tuscany/java/sca/modules/endpoint-tribes/src/main/java/org/apache/tuscany/sca/endpoint/tribes/AbstractReplicatedMap.java
(original)
+++
tuscany/java/sca/modules/endpoint-tribes/src/main/java/org/apache/tuscany/sca/endpoint/tribes/AbstractReplicatedMap.java
Mon Oct 12 19:31:48 2009
@@ -685,7 +685,8 @@
}
public void mapMemberAdded(Member member) {
- if (member.equals(getChannel().getLocalMember(false)))
+ Member self = getChannel().getLocalMember(false);
+ if (member.equals(self))
return;
boolean memberAdded = false;
//select a backup node if we don't have one
@@ -703,11 +704,13 @@
MapEntry entry = (MapEntry)super.get(e.getKey());
if (entry == null)
continue;
- if (entry.isPrimary() && (entry.getBackupNodes() == null
|| entry.getBackupNodes().length == 0)) {
+ // if (entry.isPrimary() && (entry.getBackupNodes() ==
null || entry.getBackupNodes().length == 0)) {
+ // [rfeng] Change the behavior to replicate to all nodes
+ if (entry.isPrimary() && self.equals(entry.getPrimary())) {
try {
- Member[] backup = publishEntryInfo(entry.getKey(),
entry.getValue());
+ Member[] backup = publishEntryInfo(entry.getKey(),
entry.getValue(), entry.getBackupNodes());
entry.setBackupNodes(backup);
- entry.setPrimary(channel.getLocalMember(false));
+ entry.setPrimary(self);
} catch (ChannelException x) {
log.error("Unable to select backup node.", x);
} //catch
@@ -765,7 +768,7 @@
if (log.isDebugEnabled())
log.debug("[1] Primary choosing a new backup");
try {
- Member[] backup = publishEntryInfo(entry.getKey(),
entry.getValue());
+ Member[] backup = publishEntryInfo(entry.getKey(),
entry.getValue(), entry.getBackupNodes());
entry.setBackupNodes(backup);
entry.setPrimary(channel.getLocalMember(false));
} catch (ChannelException x) {
@@ -795,7 +798,7 @@
entry.setPrimary(channel.getLocalMember(false));
entry.setBackup(false);
entry.setProxy(false);
- Member[] backup = publishEntryInfo(entry.getKey(),
entry.getValue());
+ Member[] backup = publishEntryInfo(entry.getKey(),
entry.getValue(), entry.getBackupNodes());
entry.setBackupNodes(backup);
if (mapOwner != null)
mapOwner.objectMadePrimay(entry.getKey(),
entry.getValue());
@@ -830,7 +833,7 @@
return members[node];
}
- protected abstract Member[] publishEntryInfo(Object key, Object value)
throws ChannelException;
+ protected abstract Member[] publishEntryInfo(Object key, Object value,
Member[] backupNodes) throws ChannelException;
public void heartbeat() {
try {
@@ -913,7 +916,7 @@
}
if (entry.isBackup()) {
//select a new backup node
- backup = publishEntryInfo(key, entry.getValue());
+ backup = publishEntryInfo(key, entry.getValue(),
entry.getBackupNodes());
} else if (entry.isProxy()) {
//invalidate the previous primary
msg =
@@ -994,7 +997,7 @@
old = remove(key);
try {
if (notify) {
- Member[] backup = publishEntryInfo(key, value);
+ Member[] backup = publishEntryInfo(key, value,
entry.getBackupNodes());
entry.setBackupNodes(backup);
}
} catch (ChannelException x) {
Modified:
tuscany/java/sca/modules/endpoint-tribes/src/main/java/org/apache/tuscany/sca/endpoint/tribes/ReplicatedEndpointRegistry.java
URL:
http://svn.apache.org/viewvc/tuscany/java/sca/modules/endpoint-tribes/src/main/java/org/apache/tuscany/sca/endpoint/tribes/ReplicatedEndpointRegistry.java?rev=824468&r1=824467&r2=824468&view=diff
==============================================================================
---
tuscany/java/sca/modules/endpoint-tribes/src/main/java/org/apache/tuscany/sca/endpoint/tribes/ReplicatedEndpointRegistry.java
(original)
+++
tuscany/java/sca/modules/endpoint-tribes/src/main/java/org/apache/tuscany/sca/endpoint/tribes/ReplicatedEndpointRegistry.java
Mon Oct 12 19:31:48 2009
@@ -75,6 +75,8 @@
private ExtensionPointRegistry registry;
private ReplicatedMap map;
private static List<URL> staticRoutes;
+
+ private String id;
private static final Channel createChannel(String address, int port,
String bindAddress) {
@@ -103,7 +105,7 @@
String domainURI) {
this.registry = registry;
this.domainURI = domainURI;
-
+ this.id = "[" + System.identityHashCode(this) + "]";
getParameters(domainRegistryURI);
}
@@ -207,7 +209,7 @@
try {
channel.stop(Channel.DEFAULT);
} catch (ChannelException e) {
- throw new IllegalStateException(e);
+ logger.log(Level.WARNING, e.getMessage(), e);
}
map = null;
}
@@ -336,6 +338,10 @@
public void removeListener(EndpointListener listener) {
listeners.remove(listener);
}
+
+ public void replicate(boolean complete) {
+ map.replicate(complete);
+ }
public void updateEndpoint(String uri, Endpoint endpoint) {
Endpoint oldEndpoint = getEndpoint(uri);
@@ -348,7 +354,7 @@
public void entryAdded(Object key, Object value) {
MapEntry entry = (MapEntry)value;
if (!isLocal(entry)) {
- logger.info("Remote endpoint added: " + entry.getValue());
+ logger.info(id + " Remote endpoint added: " + entry.getValue());
}
Endpoint newEp = (Endpoint)entry.getValue();
for (EndpointListener listener : listeners) {
@@ -359,7 +365,7 @@
public void entryRemoved(Object key, Object value) {
MapEntry entry = (MapEntry)value;
if (!isLocal(entry)) {
- logger.info("Remote endpoint removed: " + entry.getValue());
+ logger.info(id + " Remote endpoint removed: " + entry.getValue());
}
Endpoint oldEp = (Endpoint)entry.getValue();
for (EndpointListener listener : listeners) {
@@ -371,7 +377,7 @@
MapEntry oldEntry = (MapEntry)oldValue;
MapEntry newEntry = (MapEntry)newValue;
if (!isLocal(newEntry)) {
- logger.info("Remote endpoint updated: " + newEntry.getValue());
+ logger.info(id + " Remote endpoint updated: " +
newEntry.getValue());
}
Endpoint oldEp = (Endpoint)oldEntry.getValue();
Endpoint newEp = (Endpoint)newEntry.getValue();
Modified:
tuscany/java/sca/modules/endpoint-tribes/src/main/java/org/apache/tuscany/sca/endpoint/tribes/ReplicatedMap.java
URL:
http://svn.apache.org/viewvc/tuscany/java/sca/modules/endpoint-tribes/src/main/java/org/apache/tuscany/sca/endpoint/tribes/ReplicatedMap.java?rev=824468&r1=824467&r2=824468&view=diff
==============================================================================
---
tuscany/java/sca/modules/endpoint-tribes/src/main/java/org/apache/tuscany/sca/endpoint/tribes/ReplicatedMap.java
(original)
+++
tuscany/java/sca/modules/endpoint-tribes/src/main/java/org/apache/tuscany/sca/endpoint/tribes/ReplicatedMap.java
Mon Oct 12 19:31:48 2009
@@ -97,7 +97,7 @@
* @return Member - the backup node
* @throws ChannelException
*/
- protected Member[] publishEntryInfo(Object key, Object value) throws
ChannelException {
+ protected Member[] publishEntryInfo(Object key, Object value, Member[]
backupNodes) throws ChannelException {
if (!(key instanceof Serializable && value instanceof Serializable))
return new Member[0];
//select a backup node
@@ -106,12 +106,18 @@
if (backup == null || backup.length == 0)
return null;
+ // Set the receivers to these members that are not in the backup nodes
yet
+ Member[] members = backup;
+ if (backupNodes != null) {
+ members = getMapMembersExcl(backupNodes);
+ }
+
//publish the data out to all nodes
MapMessage msg =
new MapMessage(getMapContextName(), MapMessage.MSG_COPY, false,
(Serializable)key, (Serializable)value,
null, channel.getLocalMember(false), backup);
- getChannel().send(getMapMembers(), msg, getChannelSendOptions());
+ getChannel().send(members, msg, getChannelSendOptions());
return backup;
}
Copied:
tuscany/java/sca/modules/endpoint-tribes/src/test/java/org/apache/tuscany/sca/endpoint/tribes/ReplicatedEndpointRegistryTestCase.java
(from r824393,
tuscany/java/sca/modules/endpoint-tribes/src/test/java/org/apache/tuscany/sca/endpoint/tribes/Test.java)
URL:
http://svn.apache.org/viewvc/tuscany/java/sca/modules/endpoint-tribes/src/test/java/org/apache/tuscany/sca/endpoint/tribes/ReplicatedEndpointRegistryTestCase.java?p2=tuscany/java/sca/modules/endpoint-tribes/src/test/java/org/apache/tuscany/sca/endpoint/tribes/ReplicatedEndpointRegistryTestCase.java&p1=tuscany/java/sca/modules/endpoint-tribes/src/test/java/org/apache/tuscany/sca/endpoint/tribes/Test.java&r1=824393&r2=824468&rev=824468&view=diff
==============================================================================
---
tuscany/java/sca/modules/endpoint-tribes/src/test/java/org/apache/tuscany/sca/endpoint/tribes/Test.java
(original)
+++
tuscany/java/sca/modules/endpoint-tribes/src/test/java/org/apache/tuscany/sca/endpoint/tribes/ReplicatedEndpointRegistryTestCase.java
Mon Oct 12 19:31:48 2009
@@ -23,42 +23,56 @@
import org.apache.tuscany.sca.assembly.Endpoint;
import org.apache.tuscany.sca.core.DefaultExtensionPointRegistry;
import org.apache.tuscany.sca.core.FactoryExtensionPoint;
+import org.junit.Assert;
+import org.junit.Ignore;
+import org.junit.Test;
-public class Test {
+public class ReplicatedEndpointRegistryTestCase {
- public static void main(String[] args) throws InterruptedException {
+ @Test
+ @Ignore("Ignore this test case for now as it might be sensitive to the
multicast settings for a multi-homed machine")
+ public void testReplicate() throws InterruptedException {
DefaultExtensionPointRegistry extensionPoints = new
DefaultExtensionPointRegistry();
FactoryExtensionPoint factories =
extensionPoints.getExtensionPoint(FactoryExtensionPoint.class);
AssemblyFactory assemblyFactory =
factories.getFactory(AssemblyFactory.class);
-
+
ReplicatedEndpointRegistry ep1 = new
ReplicatedEndpointRegistry(extensionPoints, null, "foo", "bar");
System.out.println("ep1 is: " + ep1);
ep1.start();
-
+
Endpoint e1 = assemblyFactory.createEndpoint();
e1.setURI("e1uri");
- e1.setExtensionPointRegistry(null);
+ e1.setExtensionPointRegistry(extensionPoints);
ep1.addEndpoint(e1);
-
+
Endpoint e1p = ep1.getEndpoint("e1uri");
- System.out.println(e1p);
-
+ System.out.println("EP1 in Registry 1: " + e1p);
+ Assert.assertNotNull(e1p);
+
ReplicatedEndpointRegistry ep2 = new
ReplicatedEndpointRegistry(extensionPoints, null, "foo", "bar");
System.out.println("ep2 is: " + ep2);
ep2.start();
- Thread.sleep(10000);
-
+ Thread.sleep(5000);
+
Endpoint e1p2 = ep2.getEndpoint("e1uri");
- System.out.println(e1p2);
-
-
+ System.out.println("EP1 in Registry 2: " + e1p2);
+ Assert.assertNotNull(e1p2);
+
ReplicatedEndpointRegistry ep3 = new
ReplicatedEndpointRegistry(extensionPoints, null, "foo", "bar");
System.out.println("ep3 is: " + ep3);
ep3.start();
- Thread.sleep(10000);
-
+ Thread.sleep(5000);
+
Endpoint e1p3 = ep3.getEndpoint("e1uri");
- System.out.println(e1p3);
-
+ System.out.println("EP1 in Registry 3: " + e1p3);
+ Assert.assertNotNull(e1p3);
+
+ ep1.stop();
+ ep2.stop();
+ ep3.stop();
+ }
+
+ public static void main(String[] args) throws Exception {
+ new ReplicatedEndpointRegistryTestCase().testReplicate();
}
}
