rdev/gossip-revisited
1
1
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

WIP on configurable cluster names

Browse Source
This commit is contained in:
P. Taylor Goetz
2016-04-26 17:08:56 -04:00
parent bea44ea12e
commit 723c0a050b
13 changed files with 71 additions and 32 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
src/main/java/com/google/code/gossip/manager/GossipManager.java
View File

@ -57,17 +57,17 @@ public abstract class GossipManager extends Thread implements NotificationListen
private ExecutorService _gossipThreadExecutor;
public GossipManager(Class<? extends PassiveGossipThread> passiveGossipThreadClass,
Class<? extends ActiveGossipThread> activeGossipThreadClass, String address, int port,
Class<? extends ActiveGossipThread> activeGossipThreadClass, String cluster, String address, int port,
String id, GossipSettings settings, List<GossipMember> gossipMembers,
GossipListener listener) {
_passiveGossipThreadClass = passiveGossipThreadClass;
_activeGossipThreadClass = activeGossipThreadClass;
_settings = settings;
_me = new LocalGossipMember(address, port, id, System.currentTimeMillis(), this, settings.getCleanupInterval());
_me = new LocalGossipMember(cluster, address, port, id, System.currentTimeMillis(), this, settings.getCleanupInterval());
members = new ConcurrentSkipListMap<>();
for (GossipMember startupMember : gossipMembers) {
if (!startupMember.equals(_me)) {
LocalGossipMember member = new LocalGossipMember(startupMember.getHost(),
LocalGossipMember member = new LocalGossipMember(startupMember.getClusterName(), startupMember.getHost(),
startupMember.getPort(), startupMember.getId(), System.currentTimeMillis(), this,
settings.getCleanupInterval());
members.put(member, GossipState.UP);

10
src/main/java/com/google/code/gossip/manager/PassiveGossipThread.java
View File

@ -53,6 +53,8 @@ abstract public class PassiveGossipThread implements Runnable {
private AtomicBoolean _keepRunning;
private final String _cluster;
public PassiveGossipThread(GossipManager gossipManager) {
_gossipManager = gossipManager;
try {
@ -62,6 +64,7 @@ abstract public class PassiveGossipThread implements Runnable {
GossipService.LOGGER.debug("Gossip service successfully initialized on port "
+ _gossipManager.getMyself().getPort());
GossipService.LOGGER.debug("I am " + _gossipManager.getMyself());
_cluster = _gossipManager.getMyself().getClusterName();
} catch (SocketException ex) {
GossipService.LOGGER.warn(ex);
_server = null;
@ -96,8 +99,9 @@ abstract public class PassiveGossipThread implements Runnable {
JSONArray jsonArray = new JSONArray(receivedMessage);
for (int i = 0; i < jsonArray.length(); i++) {
JSONObject memberJSONObject = jsonArray.getJSONObject(i);
if (memberJSONObject.length() == 4) {
if (memberJSONObject.length() == 5 && _cluster.equals(memberJSONObject.get(GossipMember.JSON_CLUSTER))) {
RemoteGossipMember member = new RemoteGossipMember(
memberJSONObject.getString(GossipMember.JSON_CLUSTER),
memberJSONObject.getString(GossipMember.JSON_HOST),
memberJSONObject.getInt(GossipMember.JSON_PORT),
memberJSONObject.getString(GossipMember.JSON_ID),
@ -109,9 +113,11 @@ abstract public class PassiveGossipThread implements Runnable {
senderMember = member;
}
remoteGossipMembers.add(member);
} else if(memberJSONObject.length() == 5) {
GossipService.LOGGER.warn("The member object does not belong to this cluster.");
} else {
GossipService.LOGGER
.error("The received member object does not contain 4 objects:\n"
.error("The received member object does not contain 5 objects:\n"
+ memberJSONObject.toString());
}

6
src/main/java/com/google/code/gossip/manager/impl/OnlyProcessReceivedPassiveGossipThread.java
View File

@ -48,7 +48,7 @@ public class OnlyProcessReceivedPassiveGossipThread extends PassiveGossipThread
for (LocalGossipMember i : gossipManager.getDeadList()){
if (i.getId().equals(senderMember.getId())){
System.out.println(gossipManager.getMyself() +" caught a live one!");
LocalGossipMember newLocalMember = new LocalGossipMember(senderMember.getHost(),
LocalGossipMember newLocalMember = new LocalGossipMember(senderMember.getClusterName(), senderMember.getHost(),
senderMember.getPort(), senderMember.getId(), senderMember.getHeartbeat(),
gossipManager, gossipManager.getSettings().getCleanupInterval());
gossipManager.revivieMember(newLocalMember);
@ -68,7 +68,7 @@ public class OnlyProcessReceivedPassiveGossipThread extends PassiveGossipThread
}
} else if (!gossipManager.getMemberList().contains(remoteMember)
&& !gossipManager.getDeadList().contains(remoteMember) ){
LocalGossipMember newLocalMember = new LocalGossipMember(remoteMember.getHost(),
LocalGossipMember newLocalMember = new LocalGossipMember(remoteMember.getClusterName(), remoteMember.getHost(),
remoteMember.getPort(), remoteMember.getId(), remoteMember.getHeartbeat(),
gossipManager, gossipManager.getSettings().getCleanupInterval());
gossipManager.createOrRevivieMember(newLocalMember);
@ -78,7 +78,7 @@ public class OnlyProcessReceivedPassiveGossipThread extends PassiveGossipThread
LocalGossipMember localDeadMember = gossipManager.getDeadList().get(
gossipManager.getDeadList().indexOf(remoteMember));
if (remoteMember.getHeartbeat() > localDeadMember.getHeartbeat()) {
LocalGossipMember newLocalMember = new LocalGossipMember(remoteMember.getHost(),
LocalGossipMember newLocalMember = new LocalGossipMember(remoteMember.getClusterName(), remoteMember.getHost(),
remoteMember.getPort(), remoteMember.getId(), remoteMember.getHeartbeat(),
gossipManager, gossipManager.getSettings().getCleanupInterval());
gossipManager.revivieMember(newLocalMember);

4
src/main/java/com/google/code/gossip/manager/random/RandomGossipManager.java
View File

@ -26,9 +26,9 @@ import com.google.code.gossip.manager.impl.OnlyProcessReceivedPassiveGossipThrea
import java.util.List;
public class RandomGossipManager extends GossipManager {
public RandomGossipManager(String address, int port, String id, GossipSettings settings,
public RandomGossipManager(String cluster, String address, int port, String id, GossipSettings settings,
List<GossipMember> gossipMembers, GossipListener listener) {
super(OnlyProcessReceivedPassiveGossipThread.class, RandomActiveGossipThread.class, address,
super(OnlyProcessReceivedPassiveGossipThread.class, RandomActiveGossipThread.class, cluster, address,
port, id, settings, gossipMembers, listener);
}
}