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

Merge pull request #2 from edwardcapriolo/dos2unix

Browse Source 
Dos2unix
This commit is contained in:
edwardcapriolo
2015-01-17 18:24:42 -05:00
parent a96ae21269 17f1ad6f4f
commit 2585d06499
17 changed files with 1401 additions and 1307 deletions
Download Patch File Download Diff File Expand all files Collapse all files

298
src/main/java/com/google/code/gossip/GossipMember.java
View File

@ -1,149 +1,149 @@
package com.google.code.gossip; package com.google.code.gossip;
import java.net.InetSocketAddress; import java.net.InetSocketAddress;
import org.json.JSONException; import org.json.JSONException;
import org.json.JSONObject; import org.json.JSONObject;
/** /**
* A abstract class representing a gossip member. * A abstract class representing a gossip member.
* *
* @author joshclemm, harmenw * @author joshclemm, harmenw
*/ */
public abstract class GossipMember { public abstract class GossipMember {
/** The JSON key for the host property. */ /** The JSON key for the host property. */
public static final String JSON_HOST = "host"; public static final String JSON_HOST = "host";
/** The JSON key for the port property. */ /** The JSON key for the port property. */
public static final String JSON_PORT = "port"; public static final String JSON_PORT = "port";
/** The JSON key for the heartbeat property. */ /** The JSON key for the heartbeat property. */
public static final String JSON_HEARTBEAT = "heartbeat"; public static final String JSON_HEARTBEAT = "heartbeat";
public static final String JSON_ID = "id"; public static final String JSON_ID = "id";
/** The hostname or IP address of this gossip member. */ /** The hostname or IP address of this gossip member. */
protected String _host; protected String _host;
/** The port number of this gossip member. */ /** The port number of this gossip member. */
protected int _port; protected int _port;
/** The current heartbeat of this gossip member. */ /** The current heartbeat of this gossip member. */
protected int _heartbeat; protected int _heartbeat;
protected String _id; protected String _id;
/** /**
* Constructor. * Constructor.
* @param host The hostname or IP address. * @param host The hostname or IP address.
* @param port The port number. * @param port The port number.
* @param heartbeat The current heartbeat. * @param heartbeat The current heartbeat.
*/ */
public GossipMember(String host, int port, String id, int heartbeat) { public GossipMember(String host, int port, String id, int heartbeat) {
_host = host; _host = host;
_port = port; _port = port;
_id = id; _id = id;
_heartbeat = heartbeat; _heartbeat = heartbeat;
} }
/** /**
* Get the hostname or IP address of the remote gossip member. * Get the hostname or IP address of the remote gossip member.
* @return The hostname or IP address. * @return The hostname or IP address.
*/ */
public String getHost() { public String getHost() {
return _host; return _host;
} }
/** /**
* Get the port number of the remote gossip member. * Get the port number of the remote gossip member.
* @return The port number. * @return The port number.
*/ */
public int getPort() { public int getPort() {
return _port; return _port;
} }
/** /**
* The member address in the form IP/host:port * The member address in the form IP/host:port
* Similar to the toString in {@link InetSocketAddress} * Similar to the toString in {@link InetSocketAddress}
*/ */
public String getAddress() { public String getAddress() {
return _host+":"+_port; return _host+":"+_port;
} }
/** /**
* Get the heartbeat of this gossip member. * Get the heartbeat of this gossip member.
* @return The current heartbeat. * @return The current heartbeat.
*/ */
public int getHeartbeat() { public int getHeartbeat() {
return _heartbeat; return _heartbeat;
} }
/** /**
* Set the heartbeat of this gossip member. * Set the heartbeat of this gossip member.
* @param heartbeat The new heartbeat. * @param heartbeat The new heartbeat.
*/ */
public void setHeartbeat(int heartbeat) { public void setHeartbeat(int heartbeat) {
this._heartbeat = heartbeat; this._heartbeat = heartbeat;
} }
public String getId() { public String getId() {
return _id; return _id;
} }
public void setId(String _id) { public void setId(String _id) {
this._id = _id; this._id = _id;
} }
public String toString() { public String toString() {
return "Member [address=" + getAddress() + ", id=" + _id + ", heartbeat=" + _heartbeat + "]"; return "Member [address=" + getAddress() + ", id=" + _id + ", heartbeat=" + _heartbeat + "]";
} }
/** /**
* @see java.lang.Object#hashCode() * @see java.lang.Object#hashCode()
*/ */
@Override @Override
public int hashCode() { public int hashCode() {
final int prime = 31; final int prime = 31;
int result = 1; int result = 1;
String address = getAddress(); String address = getAddress();
result = prime * result result = prime * result
+ ((address == null) ? 0 : address.hashCode()); + ((address == null) ? 0 : address.hashCode());
return result; return result;
} }
/** /**
* @see java.lang.Object#equals(java.lang.Object) * @see java.lang.Object#equals(java.lang.Object)
*/ */
@Override @Override
public boolean equals(Object obj) { public boolean equals(Object obj) {
if (this == obj) { if (this == obj) {
return true; return true;
} }
if (obj == null) { if (obj == null) {
System.err.println("equals(): obj is null."); System.err.println("equals(): obj is null.");
return false; return false;
} }
if (! (obj instanceof GossipMember) ) { if (! (obj instanceof GossipMember) ) {
System.err.println("equals(): obj is not of type GossipMember."); System.err.println("equals(): obj is not of type GossipMember.");
return false; return false;
} }
// The object is the same of they both have the same address (hostname and port). // The object is the same of they both have the same address (hostname and port).
return getAddress().equals(((LocalGossipMember) obj).getAddress()); return getAddress().equals(((LocalGossipMember) obj).getAddress());
} }
/** /**
* Get the JSONObject which is the JSON representation of this GossipMember. * Get the JSONObject which is the JSON representation of this GossipMember.
* @return The JSONObject of this GossipMember. * @return The JSONObject of this GossipMember.
*/ */
public JSONObject toJSONObject() { public JSONObject toJSONObject() {
try { try {
JSONObject jsonObject = new JSONObject(); JSONObject jsonObject = new JSONObject();
jsonObject.put(JSON_HOST, _host); jsonObject.put(JSON_HOST, _host);
jsonObject.put(JSON_PORT, _port); jsonObject.put(JSON_PORT, _port);
jsonObject.put(JSON_ID, _id); jsonObject.put(JSON_ID, _id);
jsonObject.put(JSON_HEARTBEAT, _heartbeat); jsonObject.put(JSON_HEARTBEAT, _heartbeat);
return jsonObject; return jsonObject;
} catch (JSONException e) { } catch (JSONException e) {
throw new RuntimeException(e); throw new RuntimeException(e);
} }
} }
} }

93
src/main/java/com/google/code/gossip/GossipRunner.java
View File

@ -1,46 +1,47 @@
package com.google.code.gossip; package com.google.code.gossip;
import java.io.File; import java.io.File;
import java.io.FileNotFoundException; import java.io.FileNotFoundException;
import java.io.IOException; import java.io.IOException;
import org.json.JSONException; import org.json.JSONException;
public class GossipRunner { public class GossipRunner {
private StartupSettings _settings; private StartupSettings _settings;
public static void main(String[] args) { public static void main(String[] args) {
File configFile; File configFile;
if (args.length == 1) { if (args.length == 1) {
configFile = new File("./"+args[0]); configFile = new File("./" + args[0]);
} else { } else {
configFile = new File("gossip.conf"); configFile = new File("gossip.conf");
} }
new GossipRunner(configFile); new GossipRunner(configFile);
} }
public GossipRunner(File configFile) { public GossipRunner(File configFile) {
if (configFile != null && configFile.exists()) { if (configFile != null && configFile.exists()) {
try { try {
System.out.println("Parsing the configuration file..."); System.out.println("Parsing the configuration file...");
_settings = StartupSettings.fromJSONFile(configFile); _settings = StartupSettings.fromJSONFile(configFile);
GossipService gossipService = new GossipService(_settings); GossipService gossipService = new GossipService(_settings);
System.out.println("Gossip service successfully inialized, let's start it..."); System.out.println("Gossip service successfully inialized, let's start it...");
gossipService.start(); gossipService.start();
} catch (FileNotFoundException e) { } catch (FileNotFoundException e) {
System.err.println("The given file is not found!"); System.err.println("The given file is not found!");
} catch (JSONException e) { } catch (JSONException e) {
System.err.println("The given file is not in the correct JSON format!"); System.err.println("The given file is not in the correct JSON format!");
} catch (IOException e) { } catch (IOException e) {
System.err.println("Could not read the configuration file: " + e.getMessage()); System.err.println("Could not read the configuration file: " + e.getMessage());
} catch (InterruptedException e) { } catch (InterruptedException e) {
System.err.println("Error while starting the gossip service: " + e.getMessage()); System.err.println("Error while starting the gossip service: " + e.getMessage());
} }
} else { } else {
System.out.println("The gossip.conf file is not found.\n\nEither specify the path to the startup settings file or place the gossip.json file in the same folder as the JAR file."); System.out
} .println("The gossip.conf file is not found.\n\nEither specify the path to the startup settings file or place the gossip.json file in the same folder as the JAR file.");
} }
} }
}

123
src/main/java/com/google/code/gossip/GossipService.java
View File

@ -1,58 +1,65 @@
package com.google.code.gossip; package com.google.code.gossip;
import java.net.InetAddress; import java.net.InetAddress;
import java.net.SocketException; import java.net.SocketException;
import java.net.UnknownHostException; import java.net.UnknownHostException;
import java.util.ArrayList; import java.util.ArrayList;
import org.apache.log4j.Logger; import org.apache.log4j.Logger;
import com.google.code.gossip.manager.GossipManager; import com.google.code.gossip.manager.GossipManager;
import com.google.code.gossip.manager.random.RandomGossipManager; import com.google.code.gossip.manager.random.RandomGossipManager;
/** /**
* This object represents the service which is responsible for gossiping with other gossip members. * This object represents the service which is responsible for gossiping with other gossip members.
* *
* @author joshclemm, harmenw * @author joshclemm, harmenw
*/ */
public class GossipService { public class GossipService {
public static final Logger LOGGER = Logger.getLogger(GossipService.class); public static final Logger LOGGER = Logger.getLogger(GossipService.class);
private GossipManager _gossipManager;
private GossipManager _gossipManager;
/**
* Constructor with the default settings. /**
* @throws InterruptedException * Constructor with the default settings.
* @throws UnknownHostException *
*/ * @throws InterruptedException
public GossipService(StartupSettings startupSettings) throws InterruptedException, UnknownHostException { * @throws UnknownHostException
this(InetAddress.getLocalHost().getHostAddress(), startupSettings.getPort(), "", startupSettings.getLogLevel(), startupSettings.getGossipMembers(), startupSettings.getGossipSettings()); */
} public GossipService(StartupSettings startupSettings) throws InterruptedException,
UnknownHostException {
/** this(InetAddress.getLocalHost().getHostAddress(), startupSettings.getPort(), "",
* Setup the client's lists, gossiping parameters, and parse the startup config file. startupSettings.getLogLevel(), startupSettings.getGossipMembers(), startupSettings
* @throws SocketException .getGossipSettings());
* @throws InterruptedException }
* @throws UnknownHostException
*/ /**
public GossipService(String ipAddress, int port, String id, int logLevel, ArrayList<GossipMember> gossipMembers, GossipSettings settings) throws InterruptedException, UnknownHostException { * Setup the client's lists, gossiping parameters, and parse the startup config file.
_gossipManager = new RandomGossipManager(ipAddress, port, id, settings, gossipMembers); *
} * @throws SocketException
* @throws InterruptedException
public void start() { * @throws UnknownHostException
_gossipManager.start(); */
} public GossipService(String ipAddress, int port, String id, int logLevel,
ArrayList<GossipMember> gossipMembers, GossipSettings settings)
public void shutdown() { throws InterruptedException, UnknownHostException {
_gossipManager.shutdown(); _gossipManager = new RandomGossipManager(ipAddress, port, id, settings, gossipMembers);
} }
public GossipManager get_gossipManager() { public void start() {
return _gossipManager; _gossipManager.start();
} }
public void set_gossipManager(GossipManager _gossipManager) { public void shutdown() {
this._gossipManager = _gossipManager; _gossipManager.shutdown();
} }
public GossipManager get_gossipManager() {
} return _gossipManager;
}
public void set_gossipManager(GossipManager _gossipManager) {
this._gossipManager = _gossipManager;
}
}

137
src/main/java/com/google/code/gossip/GossipSettings.java
View File

@ -1,64 +1,73 @@
package com.google.code.gossip; package com.google.code.gossip;
/** /**
* In this object the settings used by the GossipService are held. * In this object the settings used by the GossipService are held.
* *
* @author harmenw * @author harmenw
*/ */
public class GossipSettings { public class GossipSettings {
/** Time between gossip'ing in ms. Default is 1 second. */ /** Time between gossip'ing in ms. Default is 1 second. */
private int _gossipInterval = 1000; private int _gossipInterval = 1000;
/** Time between cleanups in ms. Default is 10 seconds. */ /** Time between cleanups in ms. Default is 10 seconds. */
private int _cleanupInterval = 10000; private int _cleanupInterval = 10000;
/** /**
* Construct GossipSettings with default settings. * Construct GossipSettings with default settings.
*/ */
public GossipSettings() {} public GossipSettings() {
}
/**
* Construct GossipSettings with given settings. /**
* @param gossipInterval The gossip interval in ms. * Construct GossipSettings with given settings.
* @param cleanupInterval The cleanup interval in ms. *
*/ * @param gossipInterval
public GossipSettings(int gossipInterval, int cleanupInterval) { * The gossip interval in ms.
_gossipInterval = gossipInterval; * @param cleanupInterval
_cleanupInterval = cleanupInterval; * The cleanup interval in ms.
} */
public GossipSettings(int gossipInterval, int cleanupInterval) {
/** _gossipInterval = gossipInterval;
* Set the gossip interval. _cleanupInterval = cleanupInterval;
* This is the time between a gossip message is send. }
* @param gossipInterval The gossip interval in ms.
*/ /**
public void setGossipTimeout(int gossipInterval) { * Set the gossip interval. This is the time between a gossip message is send.
_gossipInterval = gossipInterval; *
} * @param gossipInterval
* The gossip interval in ms.
/** */
* Set the cleanup interval. public void setGossipTimeout(int gossipInterval) {
* This is the time between the last heartbeat received from a member and when it will be marked as dead. _gossipInterval = gossipInterval;
* @param cleanupInterval The cleanup interval in ms. }
*/
public void setCleanupInterval(int cleanupInterval) { /**
_cleanupInterval = cleanupInterval; * Set the cleanup interval. This is the time between the last heartbeat received from a member
} * and when it will be marked as dead.
*
/** * @param cleanupInterval
* Get the gossip interval. * The cleanup interval in ms.
* @return The gossip interval in ms. */
*/ public void setCleanupInterval(int cleanupInterval) {
public int getGossipInterval() { _cleanupInterval = cleanupInterval;
return _gossipInterval; }
}
/**
/** * Get the gossip interval.
* Get the clean interval. *
* @return The cleanup interval. * @return The gossip interval in ms.
*/ */
public int getCleanupInterval() { public int getGossipInterval() {
return _cleanupInterval; return _gossipInterval;
} }
}
/**
* Get the clean interval.
*
* @return The cleanup interval.
*/
public int getCleanupInterval() {
return _cleanupInterval;
}
}

124
src/main/java/com/google/code/gossip/GossipTimeoutTimer.java
View File

@ -1,61 +1,63 @@
package com.google.code.gossip; package com.google.code.gossip;
import java.util.Date; import java.util.Date;
import javax.management.NotificationListener; import javax.management.NotificationListener;
import javax.management.timer.Timer; import javax.management.timer.Timer;
/** /**
* This object represents a timer for a gossip member. * This object represents a timer for a gossip member. When the timer has elapsed without being
* When the timer has elapsed without being reset in the meantime, it will inform the GossipService about this * reset in the meantime, it will inform the GossipService about this who in turn will put the
* who in turn will put the gossip member on the dead list, because it is apparantly not alive anymore. * gossip member on the dead list, because it is apparantly not alive anymore.
* *
* @author joshclemm, harmenw * @author joshclemm, harmenw
*/ */
public class GossipTimeoutTimer extends Timer { public class GossipTimeoutTimer extends Timer {
/** The amount of time this timer waits before generating a wake-up event. */ /** The amount of time this timer waits before generating a wake-up event. */
private long _sleepTime; private long _sleepTime;
/** The gossip member this timer is for. */ /** The gossip member this timer is for. */
private LocalGossipMember _source; private LocalGossipMember _source;
/** /**
* Constructor. * Constructor. Creates a reset-able timer that wakes up after millisecondsSleepTime.
* Creates a reset-able timer that wakes up after millisecondsSleepTime. *
* @param millisecondsSleepTime The time for this timer to wait before an event. * @param millisecondsSleepTime
* @param service * The time for this timer to wait before an event.
* @param member * @param service
*/ * @param member
public GossipTimeoutTimer(long millisecondsSleepTime, NotificationListener notificationListener, LocalGossipMember member) { */
super(); public GossipTimeoutTimer(long millisecondsSleepTime, NotificationListener notificationListener,
_sleepTime = millisecondsSleepTime; LocalGossipMember member) {
_source = member; super();
addNotificationListener(notificationListener, null, null); _sleepTime = millisecondsSleepTime;
} _source = member;
addNotificationListener(notificationListener, null, null);
/** }
* @see javax.management.timer.Timer#start()
*/ /**
public void start() { * @see javax.management.timer.Timer#start()
this.reset(); */
super.start(); public void start() {
} this.reset();
super.start();
/** }
* Resets timer to start counting down from original time.
*/ /**
public void reset() { * Resets timer to start counting down from original time.
removeAllNotifications(); */
setWakeupTime(_sleepTime); public void reset() {
} removeAllNotifications();
setWakeupTime(_sleepTime);
/** }
* Adds a new wake-up time for this timer.
* @param milliseconds /**
*/ * Adds a new wake-up time for this timer.
private void setWakeupTime(long milliseconds) { *
addNotification("type", "message", _source, new Date(System.currentTimeMillis()+milliseconds)); * @param milliseconds
} */
} private void setWakeupTime(long milliseconds) {
addNotification("type", "message", _source, new Date(System.currentTimeMillis() + milliseconds));
}
}

91
src/main/java/com/google/code/gossip/LocalGossipMember.java
View File

@ -1,42 +1,49 @@
package com.google.code.gossip; package com.google.code.gossip;
import javax.management.NotificationListener; import javax.management.NotificationListener;
/** /**
* This object represent a gossip member with the properties known locally. * This object represent a gossip member with the properties known locally. These objects are stored
* These objects are stored in the local list of gossip member.s * in the local list of gossip member.s
* *
* @author harmenw * @author harmenw
*/ */
public class LocalGossipMember extends GossipMember { public class LocalGossipMember extends GossipMember {
/** The timeout timer for this gossip member. */ /** The timeout timer for this gossip member. */
private transient GossipTimeoutTimer timeoutTimer; private transient GossipTimeoutTimer timeoutTimer;
/** /**
* Constructor. * Constructor.
* @param host The hostname or IP address. *
* @param port The port number. * @param host
* @param heartbeat The current heartbeat. * The hostname or IP address.
* @param gossipService The GossipService object. * @param port
* @param cleanupTimeout The cleanup timeout for this gossip member. * The port number.
*/ * @param heartbeat
public LocalGossipMember(String hostname, int port, String id, int heartbeat, NotificationListener notificationListener, int cleanupTimeout) { * The current heartbeat.
super(hostname, port, id, heartbeat); * @param gossipService
* The GossipService object.
this.timeoutTimer = new GossipTimeoutTimer(cleanupTimeout, notificationListener, this); * @param cleanupTimeout
} * The cleanup timeout for this gossip member.
*/
/** public LocalGossipMember(String hostname, int port, String id, int heartbeat,
* Start the timeout timer. NotificationListener notificationListener, int cleanupTimeout) {
*/ super(hostname, port, id, heartbeat);
public void startTimeoutTimer() {
this.timeoutTimer.start(); this.timeoutTimer = new GossipTimeoutTimer(cleanupTimeout, notificationListener, this);
} }
/** /**
* Reset the timeout timer. * Start the timeout timer.
*/ */
public void resetTimeoutTimer() { public void startTimeoutTimer() {
this.timeoutTimer.reset(); this.timeoutTimer.start();
} }
}
/**
* Reset the timeout timer.
*/
public void resetTimeoutTimer() {
this.timeoutTimer.reset();
}
}

52
src/main/java/com/google/code/gossip/LogLevel.java
View File

@ -1,25 +1,27 @@
package com.google.code.gossip; package com.google.code.gossip;
public class LogLevel { public class LogLevel {
public static final String CONFIG_ERROR = "ERROR"; public static final String CONFIG_ERROR = "ERROR";
public static final String CONFIG_INFO = "INFO"; public static final String CONFIG_INFO = "INFO";
public static final String CONFIG_DEBUG = "DEBUG"; public static final String CONFIG_DEBUG = "DEBUG";
public static final int ERROR = 1; public static final int ERROR = 1;
public static final int INFO = 2;
public static final int DEBUG = 3; public static final int INFO = 2;
public static int fromString(String logLevel) { public static final int DEBUG = 3;
if (logLevel.equals(CONFIG_ERROR))
return ERROR; public static int fromString(String logLevel) {
else if (logLevel.equals(CONFIG_INFO)) if (logLevel.equals(CONFIG_ERROR))
return INFO; return ERROR;
else if (logLevel.equals(CONFIG_DEBUG)) else if (logLevel.equals(CONFIG_INFO))
return DEBUG; return INFO;
else else if (logLevel.equals(CONFIG_DEBUG))
return INFO; return DEBUG;
} else
} return INFO;
}
}

64
src/main/java/com/google/code/gossip/RemoteGossipMember.java
View File

@ -1,28 +1,36 @@
package com.google.code.gossip; package com.google.code.gossip;
/** /**
* The object represents a gossip member with the properties as received from a remote gossip member. * The object represents a gossip member with the properties as received from a remote gossip
* * member.
* @author harmenw *
*/ * @author harmenw
public class RemoteGossipMember extends GossipMember { */
public class RemoteGossipMember extends GossipMember {
/**
* Constructor. /**
* @param host The hostname or IP address. * Constructor.
* @param port The port number. *
* @param heartbeat The current heartbeat. * @param host
*/ * The hostname or IP address.
public RemoteGossipMember(String hostname, int port, String id, int heartbeat) { * @param port
super(hostname, port, id, heartbeat); * The port number.
} * @param heartbeat
* The current heartbeat.
/** */
* Construct a RemoteGossipMember with a heartbeat of 0. public RemoteGossipMember(String hostname, int port, String id, int heartbeat) {
* @param host The hostname or IP address. super(hostname, port, id, heartbeat);
* @param port The port number. }
*/
public RemoteGossipMember(String hostname, int port, String id) { /**
super(hostname, port, id, 0); * Construct a RemoteGossipMember with a heartbeat of 0.
} *
} * @param host
* The hostname or IP address.
* @param port
* The port number.
*/
public RemoteGossipMember(String hostname, int port, String id) {
super(hostname, port, id, 0);
}
}

345
src/main/java/com/google/code/gossip/StartupSettings.java
View File

@ -1,161 +1,184 @@
package com.google.code.gossip; package com.google.code.gossip;
import java.io.BufferedReader; import java.io.BufferedReader;
import java.io.File; import java.io.File;
import java.io.FileNotFoundException; import java.io.FileNotFoundException;
import java.io.FileReader; import java.io.FileReader;
import java.io.IOException; import java.io.IOException;
import java.util.ArrayList; import java.util.ArrayList;
import org.json.JSONArray; import org.json.JSONArray;
import org.json.JSONException; import org.json.JSONException;
import org.json.JSONObject; import org.json.JSONObject;
/** /**
* This object represents the settings used when starting the gossip service. * This object represents the settings used when starting the gossip service.
* *
* @author harmenw * @author harmenw
*/ */
public class StartupSettings { public class StartupSettings {
/** The port to start the gossip service on. */ /** The port to start the gossip service on. */
private int _port; private int _port;
/** The logging level of the gossip service. */ /** The logging level of the gossip service. */
private int _logLevel; private int _logLevel;
/** The gossip settings used at startup. */ /** The gossip settings used at startup. */
private GossipSettings _gossipSettings; private GossipSettings _gossipSettings;
/** The list with gossip members to start with. */ /** The list with gossip members to start with. */
private ArrayList<GossipMember> _gossipMembers; private ArrayList<GossipMember> _gossipMembers;
/** /**
* Constructor. * Constructor.
* @param port The port to start the service on. *
*/ * @param port
public StartupSettings(int port, int logLevel) { * The port to start the service on.
this(port, logLevel, new GossipSettings()); */
} public StartupSettings(int port, int logLevel) {
this(port, logLevel, new GossipSettings());
/** }
* Constructor.
* @param port The port to start the service on. /**
*/ * Constructor.
public StartupSettings(int port, int logLevel, GossipSettings gossipSettings) { *
_port = port; * @param port
_logLevel = logLevel; * The port to start the service on.
_gossipSettings = gossipSettings; */
_gossipMembers = new ArrayList<GossipMember>(); public StartupSettings(int port, int logLevel, GossipSettings gossipSettings) {
} _port = port;
_logLevel = logLevel;
/** _gossipSettings = gossipSettings;
* Set the port of the gossip service. _gossipMembers = new ArrayList<GossipMember>();
* @param port The port for the gossip service. }
*/
public void setPort(int port) { /**
_port = port; * Set the port of the gossip service.
} *
* @param port
/** * The port for the gossip service.
* Get the port for the gossip service. */
* @return The port of the gossip service. public void setPort(int port) {
*/ _port = port;
public int getPort() { }
return _port;
} /**
* Get the port for the gossip service.
/** *
* Set the log level of the gossip service. * @return The port of the gossip service.
* @param logLevel The log level({LogLevel}). */
*/ public int getPort() {
public void setLogLevel(int logLevel) { return _port;
_logLevel = logLevel; }
}
/**
/** * Set the log level of the gossip service.
* Get the log level of the gossip service. *
* @return The log level. * @param logLevel
*/ * The log level({LogLevel}).
public int getLogLevel() { */
return _logLevel; public void setLogLevel(int logLevel) {
} _logLevel = logLevel;
}
/**
* Get the GossipSettings. /**
* @return The GossipSettings object. * Get the log level of the gossip service.
*/ *
public GossipSettings getGossipSettings() { * @return The log level.
return _gossipSettings; */
} public int getLogLevel() {
return _logLevel;
/** }
* Add a gossip member to the list of members to start with.
* @param member The member to add. /**
*/ * Get the GossipSettings.
public void addGossipMember(GossipMember member) { *
_gossipMembers.add(member); * @return The GossipSettings object.
} */
public GossipSettings getGossipSettings() {
/** return _gossipSettings;
* Get the list with gossip members. }
* @return The gossip members.
*/ /**
public ArrayList<GossipMember> getGossipMembers() { * Add a gossip member to the list of members to start with.
return _gossipMembers; *
} * @param member
* The member to add.
/** */
* Parse the settings for the gossip service from a JSON file. public void addGossipMember(GossipMember member) {
* @param jsonFile The file object which refers to the JSON config file. _gossipMembers.add(member);
* @return The StartupSettings object with the settings from the config file. }
* @throws JSONException Thrown when the file is not well-formed JSON.
* @throws FileNotFoundException Thrown when the file cannot be found. /**
* @throws IOException Thrown when reading the file gives problems. * Get the list with gossip members.
*/ *
public static StartupSettings fromJSONFile(File jsonFile) throws JSONException, FileNotFoundException, IOException { * @return The gossip members.
// Read the file to a String. */
BufferedReader br = new BufferedReader(new FileReader(jsonFile)); public ArrayList<GossipMember> getGossipMembers() {
StringBuffer buffer = new StringBuffer(); return _gossipMembers;
String line; }
while((line = br.readLine()) != null) {
buffer.append(line.trim()); /**
} * Parse the settings for the gossip service from a JSON file.
*
// Lets parse the String as JSON. * @param jsonFile
JSONObject jsonObject = new JSONArray(buffer.toString()).getJSONObject(0); * The file object which refers to the JSON config file.
* @return The StartupSettings object with the settings from the config file.
// Now get the port number. * @throws JSONException
int port = jsonObject.getInt("port"); * Thrown when the file is not well-formed JSON.
* @throws FileNotFoundException
// Get the log level from the config file. * Thrown when the file cannot be found.
int logLevel = LogLevel.fromString(jsonObject.getString("log_level")); * @throws IOException
* Thrown when reading the file gives problems.
// Get the gossip_interval from the config file. */
int gossipInterval = jsonObject.getInt("gossip_interval"); public static StartupSettings fromJSONFile(File jsonFile) throws JSONException,
FileNotFoundException, IOException {
// Get the cleanup_interval from the config file. // Read the file to a String.
int cleanupInterval = jsonObject.getInt("cleanup_interval"); BufferedReader br = new BufferedReader(new FileReader(jsonFile));
StringBuffer buffer = new StringBuffer();
System.out.println("Config [port: " + port + ", log_level: " + logLevel + ", gossip_interval: " + gossipInterval + ", cleanup_interval: " + cleanupInterval + "]"); String line;
while ((line = br.readLine()) != null) {
// Initiate the settings with the port number. buffer.append(line.trim());
StartupSettings settings = new StartupSettings(port, logLevel, new GossipSettings(gossipInterval, cleanupInterval)); }
// Now iterate over the members from the config file and add them to the settings. // Lets parse the String as JSON.
System.out.print("Config-members ["); JSONObject jsonObject = new JSONArray(buffer.toString()).getJSONObject(0);
JSONArray membersJSON = jsonObject.getJSONArray("members");
for (int i=0; i<membersJSON.length(); i++) { // Now get the port number.
JSONObject memberJSON = membersJSON.getJSONObject(i); int port = jsonObject.getInt("port");
RemoteGossipMember member = new RemoteGossipMember(memberJSON.getString("host"), memberJSON.getInt("port"), "");
settings.addGossipMember(member); // Get the log level from the config file.
System.out.print(member.getAddress()); int logLevel = LogLevel.fromString(jsonObject.getString("log_level"));
if (i < (membersJSON.length() - 1))
System.out.print(", "); // Get the gossip_interval from the config file.
} int gossipInterval = jsonObject.getInt("gossip_interval");
System.out.println("]");
// Get the cleanup_interval from the config file.
// Return the created settings object. int cleanupInterval = jsonObject.getInt("cleanup_interval");
return settings;
} System.out.println("Config [port: " + port + ", log_level: " + logLevel + ", gossip_interval: "
} + gossipInterval + ", cleanup_interval: " + cleanupInterval + "]");
// Initiate the settings with the port number.
StartupSettings settings = new StartupSettings(port, logLevel, new GossipSettings(
gossipInterval, cleanupInterval));
// Now iterate over the members from the config file and add them to the settings.
System.out.print("Config-members [");
JSONArray membersJSON = jsonObject.getJSONArray("members");
for (int i = 0; i < membersJSON.length(); i++) {
JSONObject memberJSON = membersJSON.getJSONObject(i);
RemoteGossipMember member = new RemoteGossipMember(memberJSON.getString("host"),
memberJSON.getInt("port"), "");
settings.addGossipMember(member);
System.out.print(member.getAddress());
if (i < (membersJSON.length() - 1))
System.out.print(", ");
}
System.out.println("]");
// Return the created settings object.
return settings;
}
}

158
src/main/java/com/google/code/gossip/examples/GossipExample.java
View File

@ -1,78 +1,80 @@
package com.google.code.gossip.examples; package com.google.code.gossip.examples;
import java.net.InetAddress;
import java.net.InetAddress; import java.net.UnknownHostException;
import java.net.UnknownHostException; import java.util.ArrayList;
import java.util.ArrayList;
import com.google.code.gossip.GossipMember;
import com.google.code.gossip.GossipMember; import com.google.code.gossip.GossipService;
import com.google.code.gossip.GossipService; import com.google.code.gossip.GossipSettings;
import com.google.code.gossip.GossipSettings; import com.google.code.gossip.LogLevel;
import com.google.code.gossip.LogLevel; import com.google.code.gossip.RemoteGossipMember;
import com.google.code.gossip.RemoteGossipMember;
/**
/** * This class is an example of how one could use the gossip service. Here we start multiple gossip
* This class is an example of how one could use the gossip service. * clients on this host as specified in the config file.
* Here we start multiple gossip clients on this host as specified in the config file. *
* * @author harmenw
* @author harmenw */
*/ public class GossipExample extends Thread {
public class GossipExample extends Thread { /** The number of clients to start. */
/** The number of clients to start. */ private static final int NUMBER_OF_CLIENTS = 4;
private static final int NUMBER_OF_CLIENTS = 4;
/**
/** * @param args
* @param args */
*/ public static void main(String[] args) {
public static void main(String[] args) { new GossipExample();
new GossipExample(); }
}
/**
/** * Constructor. This will start the this thread.
* Constructor. */
* This will start the this thread. public GossipExample() {
*/ start();
public GossipExample() { }
start();
} /**
* @see java.lang.Thread#run()
/** */
* @see java.lang.Thread#run() public void run() {
*/ try {
public void run() { GossipSettings settings = new GossipSettings();
try {
GossipSettings settings = new GossipSettings(); ArrayList<GossipService> clients = new ArrayList<GossipService>();
ArrayList<GossipService> clients = new ArrayList<GossipService>(); // Get my ip address.
String myIpAddress = InetAddress.getLocalHost().getHostAddress();
// Get my ip address.
String myIpAddress = InetAddress.getLocalHost().getHostAddress(); // Create the gossip members and put them in a list and give them a port number starting with
// 2000.
// Create the gossip members and put them in a list and give them a port number starting with 2000. ArrayList<GossipMember> startupMembers = new ArrayList<GossipMember>();
ArrayList<GossipMember> startupMembers = new ArrayList<GossipMember>(); for (int i = 0; i < NUMBER_OF_CLIENTS; ++i) {
for (int i=0; i<NUMBER_OF_CLIENTS; ++i) { startupMembers.add(new RemoteGossipMember(myIpAddress, 2000 + i, ""));
startupMembers.add(new RemoteGossipMember(myIpAddress, 2000+i, "")); }
}
// Lets start the gossip clients.
// Lets start the gossip clients. // Start the clients, waiting cleaning-interval + 1 second between them which will show the
// Start the clients, waiting cleaning-interval + 1 second between them which will show the dead list handling. // dead list handling.
for (GossipMember member : startupMembers) { for (GossipMember member : startupMembers) {
GossipService gossipService = new GossipService(myIpAddress, member.getPort(), "", LogLevel.DEBUG, startupMembers, settings); GossipService gossipService = new GossipService(myIpAddress, member.getPort(), "",
clients.add(gossipService); LogLevel.DEBUG, startupMembers, settings);
gossipService.start(); clients.add(gossipService);
sleep(settings.getCleanupInterval() + 1000); gossipService.start();
} sleep(settings.getCleanupInterval() + 1000);
}
// After starting all gossip clients, first wait 10 seconds and then shut them down.
sleep(10000); // After starting all gossip clients, first wait 10 seconds and then shut them down.
System.err.println("Going to shutdown all services..."); sleep(10000);
// Since they all run in the same virtual machine and share the same executor, if one is shutdown they will all stop. System.err.println("Going to shutdown all services...");
clients.get(0).shutdown(); // Since they all run in the same virtual machine and share the same executor, if one is
// shutdown they will all stop.
} catch (UnknownHostException e) { clients.get(0).shutdown();
e.printStackTrace();
} catch (InterruptedException e) { } catch (UnknownHostException e) {
e.printStackTrace(); e.printStackTrace();
} } catch (InterruptedException e) {
} e.printStackTrace();
} }
}
}

117
src/main/java/com/google/code/gossip/manager/ActiveGossipThread.java
View File

@ -1,58 +1,59 @@
package com.google.code.gossip.manager; package com.google.code.gossip.manager;
import java.util.ArrayList; import java.util.ArrayList;
import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicBoolean;
import com.google.code.gossip.GossipService; import com.google.code.gossip.GossipService;
import com.google.code.gossip.LocalGossipMember; import com.google.code.gossip.LocalGossipMember;
/** /**
* [The active thread: periodically send gossip request.] * [The active thread: periodically send gossip request.] The class handles gossiping the membership
* The class handles gossiping the membership list. * list. This information is important to maintaining a common state among all the nodes, and is
* This information is important to maintaining a common * important for detecting failures.
* state among all the nodes, and is important for detecting */
* failures. abstract public class ActiveGossipThread implements Runnable {
*/
abstract public class ActiveGossipThread implements Runnable { private GossipManager _gossipManager;
private GossipManager _gossipManager; private final AtomicBoolean _keepRunning;
private final AtomicBoolean _keepRunning; public ActiveGossipThread(GossipManager gossipManager) {
_gossipManager = gossipManager;
public ActiveGossipThread(GossipManager gossipManager) { _keepRunning = new AtomicBoolean(true);
_gossipManager = gossipManager; }
_keepRunning = new AtomicBoolean(true);
} @Override
public void run() {
@Override while (_keepRunning.get()) {
public void run() { try {
while(_keepRunning.get()) { TimeUnit.MILLISECONDS.sleep(_gossipManager.getSettings().getGossipInterval());
try { sendMembershipList(_gossipManager.getMyself(), _gossipManager.getMemberList());
TimeUnit.MILLISECONDS.sleep(_gossipManager.getSettings().getGossipInterval()); } catch (InterruptedException e) {
sendMembershipList(_gossipManager.getMyself(), _gossipManager.getMemberList()); GossipService.LOGGER.error(e);
} catch (InterruptedException e) { _keepRunning.set(false);
GossipService.LOGGER.error(e); }
_keepRunning.set(false); }
} shutdown();
} }
shutdown();
} public void shutdown() {
_keepRunning.set(false);
public void shutdown(){ }
_keepRunning.set(false);
} /**
/** * Performs the sending of the membership list, after we have incremented our own heartbeat.
* Performs the sending of the membership list, after we have */
* incremented our own heartbeat. abstract protected void sendMembershipList(LocalGossipMember me,
*/ ArrayList<LocalGossipMember> memberList);
abstract protected void sendMembershipList(LocalGossipMember me, ArrayList<LocalGossipMember> memberList);
/**
/** * Abstract method which should be implemented by a subclass. This method should return a member
* Abstract method which should be implemented by a subclass. * of the list to gossip with.
* This method should return a member of the list to gossip with. *
* @param memberList The list of members which are stored in the local list of members. * @param memberList
* @return The chosen LocalGossipMember to gossip with. * The list of members which are stored in the local list of members.
*/ * @return The chosen LocalGossipMember to gossip with.
abstract protected LocalGossipMember selectPartner(ArrayList<LocalGossipMember> memberList); */
} abstract protected LocalGossipMember selectPartner(ArrayList<LocalGossipMember> memberList);
}

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

@ -1,157 +1,162 @@
package com.google.code.gossip.manager; package com.google.code.gossip.manager;
import java.lang.reflect.InvocationTargetException; import java.lang.reflect.InvocationTargetException;
import java.util.ArrayList; import java.util.ArrayList;
import java.util.concurrent.ExecutorService; import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors; import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicBoolean;
import javax.management.Notification; import javax.management.Notification;
import javax.management.NotificationListener; import javax.management.NotificationListener;
import com.google.code.gossip.GossipMember; import com.google.code.gossip.GossipMember;
import com.google.code.gossip.GossipService; import com.google.code.gossip.GossipService;
import com.google.code.gossip.GossipSettings; import com.google.code.gossip.GossipSettings;
import com.google.code.gossip.LocalGossipMember; import com.google.code.gossip.LocalGossipMember;
public abstract class GossipManager extends Thread implements NotificationListener { public abstract class GossipManager extends Thread implements NotificationListener {
/** The maximal number of bytes the packet with the GOSSIP may be. (Default is 100 kb) */ /** The maximal number of bytes the packet with the GOSSIP may be. (Default is 100 kb) */
public static final int MAX_PACKET_SIZE = 102400; public static final int MAX_PACKET_SIZE = 102400;
/** The list of members which are in the gossip group (not including myself). */ /** The list of members which are in the gossip group (not including myself). */
private ArrayList<LocalGossipMember> _memberList; private ArrayList<LocalGossipMember> _memberList;
/** The list of members which are known to be dead. */ /** The list of members which are known to be dead. */
private ArrayList<LocalGossipMember> _deadList; private ArrayList<LocalGossipMember> _deadList;
/** The member I am representing. */ /** The member I am representing. */
private LocalGossipMember _me; private LocalGossipMember _me;
/** The settings for gossiping. */ /** The settings for gossiping. */
private GossipSettings _settings; private GossipSettings _settings;
/** A boolean whether the gossip service should keep running. */ /** A boolean whether the gossip service should keep running. */
private AtomicBoolean _gossipServiceRunning; private AtomicBoolean _gossipServiceRunning;
/** A ExecutorService used for executing the active and passive gossip threads. */ /** A ExecutorService used for executing the active and passive gossip threads. */
private ExecutorService _gossipThreadExecutor; private ExecutorService _gossipThreadExecutor;
private Class<? extends PassiveGossipThread> _passiveGossipThreadClass; private Class<? extends PassiveGossipThread> _passiveGossipThreadClass;
private PassiveGossipThread passiveGossipThread;
private PassiveGossipThread passiveGossipThread;
private Class<? extends ActiveGossipThread> _activeGossipThreadClass;
private ActiveGossipThread activeGossipThread; private Class<? extends ActiveGossipThread> _activeGossipThreadClass;
public GossipManager(Class<? extends PassiveGossipThread> passiveGossipThreadClass, private ActiveGossipThread activeGossipThread;
Class<? extends ActiveGossipThread> activeGossipThreadClass, String address, int port,
String id, GossipSettings settings, ArrayList<GossipMember> gossipMembers) { public GossipManager(Class<? extends PassiveGossipThread> passiveGossipThreadClass,
_passiveGossipThreadClass = passiveGossipThreadClass; Class<? extends ActiveGossipThread> activeGossipThreadClass, String address, int port,
_activeGossipThreadClass = activeGossipThreadClass; String id, GossipSettings settings, ArrayList<GossipMember> gossipMembers) {
_settings = settings; _passiveGossipThreadClass = passiveGossipThreadClass;
_me = new LocalGossipMember(address, port, id, 0, this, settings.getCleanupInterval()); _activeGossipThreadClass = activeGossipThreadClass;
_memberList = new ArrayList<LocalGossipMember>(); _settings = settings;
_deadList = new ArrayList<LocalGossipMember>(); _me = new LocalGossipMember(address, port, id, 0, this, settings.getCleanupInterval());
for (GossipMember startupMember : gossipMembers) { _memberList = new ArrayList<LocalGossipMember>();
if (!startupMember.equals(_me)) { _deadList = new ArrayList<LocalGossipMember>();
LocalGossipMember member = new LocalGossipMember(startupMember.getHost(), for (GossipMember startupMember : gossipMembers) {
startupMember.getPort(), startupMember.getId(), 0, this, if (!startupMember.equals(_me)) {
settings.getCleanupInterval()); LocalGossipMember member = new LocalGossipMember(startupMember.getHost(),
_memberList.add(member); startupMember.getPort(), startupMember.getId(), 0, this,
GossipService.LOGGER.debug(member); settings.getCleanupInterval());
} _memberList.add(member);
} GossipService.LOGGER.debug(member);
}
_gossipServiceRunning = new AtomicBoolean(true); }
Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() {
public void run() { _gossipServiceRunning = new AtomicBoolean(true);
GossipService.LOGGER.info("Service has been shutdown..."); Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() {
} public void run() {
})); GossipService.LOGGER.info("Service has been shutdown...");
} }
}));
/** }
* All timers associated with a member will trigger this method when it goes
* off. The timer will go off if we have not heard from this member in /**
* <code> _settings.T_CLEANUP </code> time. * All timers associated with a member will trigger this method when it goes off. The timer will
*/ * go off if we have not heard from this member in <code> _settings.T_CLEANUP </code> time.
@Override */
public void handleNotification(Notification notification, Object handback) { @Override
LocalGossipMember deadMember = (LocalGossipMember) notification.getUserData(); public void handleNotification(Notification notification, Object handback) {
GossipService.LOGGER.info("Dead member detected: " + deadMember); LocalGossipMember deadMember = (LocalGossipMember) notification.getUserData();
synchronized (this._memberList) { GossipService.LOGGER.info("Dead member detected: " + deadMember);
this._memberList.remove(deadMember); synchronized (this._memberList) {
} this._memberList.remove(deadMember);
synchronized (this._deadList) { }
this._deadList.add(deadMember); synchronized (this._deadList) {
} this._deadList.add(deadMember);
} }
}
public GossipSettings getSettings() {
return _settings; public GossipSettings getSettings() {
} return _settings;
}
/**
* Get a clone of the memberlist. /**
* @return * Get a clone of the memberlist.
*/ *
public ArrayList<LocalGossipMember> getMemberList() { * @return
return _memberList; */
} public ArrayList<LocalGossipMember> getMemberList() {
return _memberList;
public LocalGossipMember getMyself() { }
return _me;
} public LocalGossipMember getMyself() {
return _me;
public ArrayList<LocalGossipMember> getDeadList() { }
return _deadList;
} public ArrayList<LocalGossipMember> getDeadList() {
return _deadList;
/** }
* Starts the client. Specifically, start the various cycles for this protocol.
* Start the gossip thread and start the receiver thread. /**
* @throws InterruptedException * Starts the client. Specifically, start the various cycles for this protocol. Start the gossip
*/ * thread and start the receiver thread.
public void run() { *
for (LocalGossipMember member : _memberList) { * @throws InterruptedException
if (member != _me) { */
member.startTimeoutTimer(); public void run() {
} for (LocalGossipMember member : _memberList) {
} if (member != _me) {
_gossipThreadExecutor = Executors.newCachedThreadPool(); member.startTimeoutTimer();
try { }
passiveGossipThread = _passiveGossipThreadClass.getConstructor(GossipManager.class).newInstance(this); }
_gossipThreadExecutor.execute(passiveGossipThread); _gossipThreadExecutor = Executors.newCachedThreadPool();
activeGossipThread = _activeGossipThreadClass.getConstructor(GossipManager.class).newInstance(this); try {
_gossipThreadExecutor.execute(activeGossipThread); passiveGossipThread = _passiveGossipThreadClass.getConstructor(GossipManager.class)
} catch (InstantiationException | IllegalAccessException | IllegalArgumentException .newInstance(this);
| InvocationTargetException | NoSuchMethodException | SecurityException e1) { _gossipThreadExecutor.execute(passiveGossipThread);
throw new RuntimeException(e1); activeGossipThread = _activeGossipThreadClass.getConstructor(GossipManager.class)
} .newInstance(this);
GossipService.LOGGER.info("The GossipService is started."); _gossipThreadExecutor.execute(activeGossipThread);
while(_gossipServiceRunning.get()) { } catch (InstantiationException | IllegalAccessException | IllegalArgumentException
try { | InvocationTargetException | NoSuchMethodException | SecurityException e1) {
//TODO throw new RuntimeException(e1);
TimeUnit.MILLISECONDS.sleep(1); }
} catch (InterruptedException e) { GossipService.LOGGER.info("The GossipService is started.");
GossipService.LOGGER.info("The GossipClient was interrupted."); while (_gossipServiceRunning.get()) {
} try {
} // TODO
} TimeUnit.MILLISECONDS.sleep(1);
} catch (InterruptedException e) {
/** GossipService.LOGGER.info("The GossipClient was interrupted.");
* Shutdown the gossip service. }
*/ }
public void shutdown() { }
_gossipThreadExecutor.shutdown();
passiveGossipThread.shutdown(); /**
activeGossipThread.shutdown(); * Shutdown the gossip service.
try { */
boolean result = _gossipThreadExecutor.awaitTermination(1000, TimeUnit.MILLISECONDS); public void shutdown() {
System.err.println("Terminate retuned " + result); _gossipThreadExecutor.shutdown();
} catch (InterruptedException e) { passiveGossipThread.shutdown();
e.printStackTrace(); activeGossipThread.shutdown();
} try {
_gossipServiceRunning.set(false); boolean result = _gossipThreadExecutor.awaitTermination(1000, TimeUnit.MILLISECONDS);
} System.err.println("Terminate retuned " + result);
} } catch (InterruptedException e) {
e.printStackTrace();
}
_gossipServiceRunning.set(false);
}
}

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

@ -1,142 +1,148 @@
package com.google.code.gossip.manager; package com.google.code.gossip.manager;
import java.io.IOException; import java.io.IOException;
import java.net.DatagramPacket; import java.net.DatagramPacket;
import java.net.DatagramSocket; import java.net.DatagramSocket;
import java.net.InetSocketAddress; import java.net.InetSocketAddress;
import java.net.SocketAddress; import java.net.SocketAddress;
import java.net.SocketException; import java.net.SocketException;
import java.util.ArrayList; import java.util.ArrayList;
import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicBoolean;
import org.json.JSONArray; import org.json.JSONArray;
import org.json.JSONException; import org.json.JSONException;
import org.json.JSONObject; import org.json.JSONObject;
import com.google.code.gossip.GossipMember; import com.google.code.gossip.GossipMember;
import com.google.code.gossip.GossipService; import com.google.code.gossip.GossipService;
import com.google.code.gossip.RemoteGossipMember; import com.google.code.gossip.RemoteGossipMember;
/** /**
* [The passive thread: reply to incoming gossip request.] * [The passive thread: reply to incoming gossip request.] This class handles the passive cycle,
* This class handles the passive cycle, where this client * where this client has received an incoming message. For now, this message is always the
* has received an incoming message. For now, this message * membership list, but if you choose to gossip additional information, you will need some logic to
* is always the membership list, but if you choose to gossip * determine the incoming message.
* additional information, you will need some logic to determine */
* the incoming message. abstract public class PassiveGossipThread implements Runnable {
*/
abstract public class PassiveGossipThread implements Runnable { /** The socket used for the passive thread of the gossip service. */
private DatagramSocket _server;
/** The socket used for the passive thread of the gossip service. */
private DatagramSocket _server; private GossipManager _gossipManager;
private GossipManager _gossipManager; private AtomicBoolean _keepRunning;
private AtomicBoolean _keepRunning; public PassiveGossipThread(GossipManager gossipManager) {
_gossipManager = gossipManager;
public PassiveGossipThread(GossipManager gossipManager) { try {
_gossipManager = gossipManager; SocketAddress socketAddress = new InetSocketAddress(_gossipManager.getMyself().getHost(),
try { _gossipManager.getMyself().getPort());
SocketAddress socketAddress = new InetSocketAddress(_gossipManager.getMyself().getHost(), _gossipManager.getMyself().getPort()); _server = new DatagramSocket(socketAddress);
_server = new DatagramSocket(socketAddress); GossipService.LOGGER.info("Gossip service successfully initialized on port "
GossipService.LOGGER.info("Gossip service successfully initialized on port " + _gossipManager.getMyself().getPort()); + _gossipManager.getMyself().getPort());
GossipService.LOGGER.debug("I am " + _gossipManager.getMyself()); GossipService.LOGGER.debug("I am " + _gossipManager.getMyself());
} catch (SocketException ex) { } catch (SocketException ex) {
GossipService.LOGGER.error(ex); GossipService.LOGGER.error(ex);
_server = null; _server = null;
throw new RuntimeException(ex); throw new RuntimeException(ex);
} }
_keepRunning = new AtomicBoolean(true); _keepRunning = new AtomicBoolean(true);
} }
@Override @Override
public void run() { public void run() {
while (_keepRunning.get()) { while (_keepRunning.get()) {
try { try {
byte[] buf = new byte[_server.getReceiveBufferSize()]; byte[] buf = new byte[_server.getReceiveBufferSize()];
DatagramPacket p = new DatagramPacket(buf, buf.length); DatagramPacket p = new DatagramPacket(buf, buf.length);
_server.receive(p); _server.receive(p);
GossipService.LOGGER.debug("A message has been received from " + p.getAddress() + ":" GossipService.LOGGER.debug("A message has been received from " + p.getAddress() + ":"
+ p.getPort() + "."); + p.getPort() + ".");
int packet_length = 0; int packet_length = 0;
for (int i = 0; i < 4; i++) { for (int i = 0; i < 4; i++) {
int shift = (4 - 1 - i) * 8; int shift = (4 - 1 - i) * 8;
packet_length += (buf[i] & 0x000000FF) << shift; packet_length += (buf[i] & 0x000000FF) << shift;
} }
// Check whether the package is smaller than the maximal packet length. // Check whether the package is smaller than the maximal packet length.
// A package larger than this would not be possible to be send from a GossipService, // A package larger than this would not be possible to be send from a GossipService,
// since this is check before sending the message. // since this is check before sending the message.
// This could normally only occur when the list of members is very big, // This could normally only occur when the list of members is very big,
// or when the packet is misformed, and the first 4 bytes is not the right in anymore. // or when the packet is misformed, and the first 4 bytes is not the right in anymore.
// For this reason we regards the message. // For this reason we regards the message.
if (packet_length <= GossipManager.MAX_PACKET_SIZE) { if (packet_length <= GossipManager.MAX_PACKET_SIZE) {
byte[] json_bytes = new byte[packet_length]; byte[] json_bytes = new byte[packet_length];
for (int i = 0; i < packet_length; i++) { for (int i = 0; i < packet_length; i++) {
json_bytes[i] = buf[i + 4]; json_bytes[i] = buf[i + 4];
} }
String receivedMessage = new String(json_bytes); String receivedMessage = new String(json_bytes);
GossipService.LOGGER.debug("Received message (" + packet_length + " bytes): " GossipService.LOGGER.debug("Received message (" + packet_length + " bytes): "
+ receivedMessage); + receivedMessage);
try { try {
ArrayList<GossipMember> remoteGossipMembers = new ArrayList<GossipMember>(); ArrayList<GossipMember> remoteGossipMembers = new ArrayList<GossipMember>();
RemoteGossipMember senderMember = null; RemoteGossipMember senderMember = null;
GossipService.LOGGER.debug("Received member list:"); GossipService.LOGGER.debug("Received member list:");
JSONArray jsonArray = new JSONArray(receivedMessage); JSONArray jsonArray = new JSONArray(receivedMessage);
for (int i = 0; i < jsonArray.length(); i++) { for (int i = 0; i < jsonArray.length(); i++) {
JSONObject memberJSONObject = jsonArray.getJSONObject(i); JSONObject memberJSONObject = jsonArray.getJSONObject(i);
if (memberJSONObject.length() == 4) { if (memberJSONObject.length() == 4) {
RemoteGossipMember member = new RemoteGossipMember( RemoteGossipMember member = new RemoteGossipMember(
memberJSONObject.getString(GossipMember.JSON_HOST), memberJSONObject.getString(GossipMember.JSON_HOST),
memberJSONObject.getInt(GossipMember.JSON_PORT), memberJSONObject.getInt(GossipMember.JSON_PORT),
memberJSONObject.getString(GossipMember.JSON_ID), memberJSONObject.getString(GossipMember.JSON_ID),
memberJSONObject.getInt(GossipMember.JSON_HEARTBEAT)); memberJSONObject.getInt(GossipMember.JSON_HEARTBEAT));
GossipService.LOGGER.debug(member.toString()); GossipService.LOGGER.debug(member.toString());
// This is the first member found, so this should be the member who is communicating // This is the first member found, so this should be the member who is communicating
// with me. // with me.
if (i == 0) { if (i == 0) {
senderMember = member; senderMember = member;
} }
remoteGossipMembers.add(member); remoteGossipMembers.add(member);
} else { } else {
GossipService.LOGGER GossipService.LOGGER
.error("The received member object does not contain 4 objects:\n" .error("The received member object does not contain 4 objects:\n"
+ memberJSONObject.toString()); + memberJSONObject.toString());
} }
} }
// Merge our list with the one we just received // Merge our list with the one we just received
mergeLists(_gossipManager, senderMember, remoteGossipMembers); mergeLists(_gossipManager, senderMember, remoteGossipMembers);
} catch (JSONException e) { } catch (JSONException e) {
GossipService.LOGGER GossipService.LOGGER
.error("The received message is not well-formed JSON. The following message has been dropped:\n" .error("The received message is not well-formed JSON. The following message has been dropped:\n"
+ receivedMessage); + receivedMessage);
} }
} else { } else {
GossipService.LOGGER GossipService.LOGGER
.error("The received message is not of the expected size, it has been dropped."); .error("The received message is not of the expected size, it has been dropped.");
} }
} catch (IOException e) { } catch (IOException e) {
e.printStackTrace(); e.printStackTrace();
_keepRunning.set(false); _keepRunning.set(false);
} }
} }
shutdown(); shutdown();
} }
public void shutdown(){ public void shutdown() {
_server.close(); _server.close();
} }
/** /**
* Abstract method for merging the local and remote list. * Abstract method for merging the local and remote list.
* @param gossipManager The GossipManager for retrieving the local members and dead members list. *
* @param senderMember The member who is sending this list, this could be used to send a response if the remote list contains out-dated information. * @param gossipManager
* @param remoteList The list of members known at the remote side. * The GossipManager for retrieving the local members and dead members list.
*/ * @param senderMember
abstract protected void mergeLists(GossipManager gossipManager, RemoteGossipMember senderMember, ArrayList<GossipMember> remoteList); * The member who is sending this list, this could be used to send a response if the
* remote list contains out-dated information.
* @param remoteList
* The list of members known at the remote side.
*/
abstract protected void mergeLists(GossipManager gossipManager, RemoteGossipMember senderMember,
ArrayList<GossipMember> remoteList);
} }

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

@ -1,93 +1,111 @@
package com.google.code.gossip.manager.impl; package com.google.code.gossip.manager.impl;
import java.util.ArrayList; import java.util.ArrayList;
import com.google.code.gossip.GossipMember; import com.google.code.gossip.GossipMember;
import com.google.code.gossip.GossipService; import com.google.code.gossip.GossipService;
import com.google.code.gossip.LocalGossipMember; import com.google.code.gossip.LocalGossipMember;
import com.google.code.gossip.RemoteGossipMember; import com.google.code.gossip.RemoteGossipMember;
import com.google.code.gossip.manager.GossipManager; import com.google.code.gossip.manager.GossipManager;
import com.google.code.gossip.manager.PassiveGossipThread; import com.google.code.gossip.manager.PassiveGossipThread;
public class OnlyProcessReceivedPassiveGossipThread extends PassiveGossipThread { public class OnlyProcessReceivedPassiveGossipThread extends PassiveGossipThread {
public OnlyProcessReceivedPassiveGossipThread(GossipManager gossipManager) { public OnlyProcessReceivedPassiveGossipThread(GossipManager gossipManager) {
super(gossipManager); super(gossipManager);
} }
/** /**
* Merge remote list (received from peer), and our local member list. * Merge remote list (received from peer), and our local member list. Simply, we must update the
* Simply, we must update the heartbeats that the remote list has with * heartbeats that the remote list has with our list. Also, some additional logic is needed to
* our list. Also, some additional logic is needed to make sure we have * make sure we have not timed out a member and then immediately received a list with that member.
* not timed out a member and then immediately received a list with that *
* member. * @param remoteList
* @param remoteList */
*/ protected void mergeLists(GossipManager gossipManager, RemoteGossipMember senderMember,
protected void mergeLists(GossipManager gossipManager, RemoteGossipMember senderMember, ArrayList<GossipMember> remoteList) { ArrayList<GossipMember> remoteList) {
synchronized (gossipManager.getDeadList()) { synchronized (gossipManager.getDeadList()) {
synchronized (gossipManager.getMemberList()) { synchronized (gossipManager.getMemberList()) {
for (GossipMember remoteMember : remoteList) { for (GossipMember remoteMember : remoteList) {
// Skip myself. We don't want ourselves in the local member list. // Skip myself. We don't want ourselves in the local member list.
if (!remoteMember.equals(gossipManager.getMyself())) { if (!remoteMember.equals(gossipManager.getMyself())) {
if (gossipManager.getMemberList().contains(remoteMember)) { if (gossipManager.getMemberList().contains(remoteMember)) {
GossipService.LOGGER.debug("The local list already contains the remote member (" + remoteMember + ")."); GossipService.LOGGER.debug("The local list already contains the remote member ("
// The local memberlist contains the remote member. + remoteMember + ").");
LocalGossipMember localMember = gossipManager.getMemberList().get(gossipManager.getMemberList().indexOf(remoteMember)); // The local memberlist contains the remote member.
LocalGossipMember localMember = gossipManager.getMemberList().get(
// Let's synchronize it's heartbeat. gossipManager.getMemberList().indexOf(remoteMember));
if (remoteMember.getHeartbeat() > localMember.getHeartbeat()) {
// update local list with latest heartbeat // Let's synchronize it's heartbeat.
localMember.setHeartbeat(remoteMember.getHeartbeat()); if (remoteMember.getHeartbeat() > localMember.getHeartbeat()) {
// and reset the timeout of that member // update local list with latest heartbeat
localMember.resetTimeoutTimer(); localMember.setHeartbeat(remoteMember.getHeartbeat());
} // and reset the timeout of that member
// TODO: Otherwise, should we inform the other when the heartbeat is already higher? localMember.resetTimeoutTimer();
} else { }
// The local list does not contain the remote member. // TODO: Otherwise, should we inform the other when the heartbeat is already higher?
GossipService.LOGGER.debug("The local list does not contain the remote member (" + remoteMember + ")."); } else {
// The local list does not contain the remote member.
// The remote member is either brand new, or a previously declared dead member. GossipService.LOGGER.debug("The local list does not contain the remote member ("
// If its dead, check the heartbeat because it may have come back from the dead. + remoteMember + ").");
if (gossipManager.getDeadList().contains(remoteMember)) {
// The remote member is known here as a dead member. // The remote member is either brand new, or a previously declared dead member.
GossipService.LOGGER.debug("The remote member is known here as a dead member."); // If its dead, check the heartbeat because it may have come back from the dead.
LocalGossipMember localDeadMember = gossipManager.getDeadList().get(gossipManager.getDeadList().indexOf(remoteMember)); if (gossipManager.getDeadList().contains(remoteMember)) {
// If a member is restarted the heartbeat will restart from 1, so we should check that here. // The remote member is known here as a dead member.
// So a member can become from the dead when it is either larger than a previous heartbeat (due to network failure) GossipService.LOGGER.debug("The remote member is known here as a dead member.");
// or when the heartbeat is 1 (after a restart of the service). LocalGossipMember localDeadMember = gossipManager.getDeadList().get(
// TODO: What if the first message of a gossip service is sent to a dead node? The second member will receive a heartbeat of two. gossipManager.getDeadList().indexOf(remoteMember));
// TODO: The above does happen. Maybe a special message for a revived member? // If a member is restarted the heartbeat will restart from 1, so we should check
// TODO: Or maybe when a member is declared dead for more than _settings.getCleanupInterval() ms, reset the heartbeat to 0. // that here.
// It will then accept a revived member. // So a member can become from the dead when it is either larger than a previous
// The above is now handle by checking whether the heartbeat differs _settings.getCleanupInterval(), it must be restarted. // heartbeat (due to network failure)
if (remoteMember.getHeartbeat() == 1 // or when the heartbeat is 1 (after a restart of the service).
|| ((localDeadMember.getHeartbeat() - remoteMember.getHeartbeat()) * -1) > (gossipManager.getSettings().getCleanupInterval() / 1000) // TODO: What if the first message of a gossip service is sent to a dead node? The
|| remoteMember.getHeartbeat() > localDeadMember.getHeartbeat()) { // second member will receive a heartbeat of two.
GossipService.LOGGER.debug("The remote member is back from the dead. We will remove it from the dead list and add it as a new member."); // TODO: The above does happen. Maybe a special message for a revived member?
// The remote member is back from the dead. // TODO: Or maybe when a member is declared dead for more than
// Remove it from the dead list. // _settings.getCleanupInterval() ms, reset the heartbeat to 0.
gossipManager.getDeadList().remove(localDeadMember); // It will then accept a revived member.
// Add it as a new member and add it to the member list. // The above is now handle by checking whether the heartbeat differs
LocalGossipMember newLocalMember = new LocalGossipMember(remoteMember.getHost(), remoteMember.getPort(), remoteMember.getId(), remoteMember.getHeartbeat(), gossipManager, gossipManager.getSettings().getCleanupInterval()); // _settings.getCleanupInterval(), it must be restarted.
gossipManager.getMemberList().add(newLocalMember); if (remoteMember.getHeartbeat() == 1
newLocalMember.startTimeoutTimer(); || ((localDeadMember.getHeartbeat() - remoteMember.getHeartbeat()) * -1) > (gossipManager
GossipService.LOGGER.info("Removed remote member " + remoteMember.getAddress() + " from dead list and added to local member list."); .getSettings().getCleanupInterval() / 1000)
} || remoteMember.getHeartbeat() > localDeadMember.getHeartbeat()) {
} else { GossipService.LOGGER
// Brand spanking new member - welcome. .debug("The remote member is back from the dead. We will remove it from the dead list and add it as a new member.");
LocalGossipMember newLocalMember = new LocalGossipMember(remoteMember.getHost(), remoteMember.getPort(), remoteMember.getId(), remoteMember.getHeartbeat(), gossipManager, gossipManager.getSettings().getCleanupInterval()); // The remote member is back from the dead.
gossipManager.getMemberList().add(newLocalMember); // Remove it from the dead list.
newLocalMember.startTimeoutTimer(); gossipManager.getDeadList().remove(localDeadMember);
GossipService.LOGGER.info("Added new remote member " + remoteMember.getAddress() + " to local member list."); // Add it as a new member and add it to the member list.
} LocalGossipMember newLocalMember = new LocalGossipMember(remoteMember.getHost(),
} remoteMember.getPort(), remoteMember.getId(),
} remoteMember.getHeartbeat(), gossipManager, gossipManager.getSettings()
} .getCleanupInterval());
} gossipManager.getMemberList().add(newLocalMember);
} newLocalMember.startTimeoutTimer();
} GossipService.LOGGER.info("Removed remote member " + remoteMember.getAddress()
+ " from dead list and added to local member list.");
} }
} else {
// Brand spanking new member - welcome.
LocalGossipMember newLocalMember = new LocalGossipMember(remoteMember.getHost(),
remoteMember.getPort(), remoteMember.getId(), remoteMember.getHeartbeat(),
gossipManager, gossipManager.getSettings().getCleanupInterval());
gossipManager.getMemberList().add(newLocalMember);
newLocalMember.startTimeoutTimer();
GossipService.LOGGER.info("Added new remote member " + remoteMember.getAddress()
+ " to local member list.");
}
}
}
}
}
}
}
}

189
src/main/java/com/google/code/gossip/manager/impl/SendMembersActiveGossipThread.java
View File

@ -1,94 +1,95 @@
package com.google.code.gossip.manager.impl; package com.google.code.gossip.manager.impl;
import java.io.IOException; import java.io.IOException;
import java.net.DatagramPacket; import java.net.DatagramPacket;
import java.net.DatagramSocket; import java.net.DatagramSocket;
import java.net.InetAddress; import java.net.InetAddress;
import java.nio.ByteBuffer; import java.nio.ByteBuffer;
import java.util.ArrayList; import java.util.ArrayList;
import org.json.JSONArray; import org.json.JSONArray;
import com.google.code.gossip.GossipService; import com.google.code.gossip.GossipService;
import com.google.code.gossip.LocalGossipMember; import com.google.code.gossip.LocalGossipMember;
import com.google.code.gossip.manager.ActiveGossipThread; import com.google.code.gossip.manager.ActiveGossipThread;
import com.google.code.gossip.manager.GossipManager; import com.google.code.gossip.manager.GossipManager;
abstract public class SendMembersActiveGossipThread extends ActiveGossipThread { abstract public class SendMembersActiveGossipThread extends ActiveGossipThread {
public SendMembersActiveGossipThread(GossipManager gossipManager) { public SendMembersActiveGossipThread(GossipManager gossipManager) {
super(gossipManager); super(gossipManager);
} }
/** /**
* Performs the sending of the membership list, after we have * Performs the sending of the membership list, after we have incremented our own heartbeat.
* incremented our own heartbeat. */
*/ protected void sendMembershipList(LocalGossipMember me, ArrayList<LocalGossipMember> memberList) {
protected void sendMembershipList(LocalGossipMember me, ArrayList<LocalGossipMember> memberList) { GossipService.LOGGER.debug("Send sendMembershipList() is called.");
GossipService.LOGGER.debug("Send sendMembershipList() is called.");
// Increase the heartbeat of myself by 1.
// Increase the heartbeat of myself by 1. me.setHeartbeat(me.getHeartbeat() + 1);
me.setHeartbeat(me.getHeartbeat() + 1);
synchronized (memberList) {
synchronized (memberList) { try {
try { LocalGossipMember member = selectPartner(memberList);
LocalGossipMember member = selectPartner(memberList);
if (member != null) {
if (member != null) { InetAddress dest = InetAddress.getByName(member.getHost());
InetAddress dest = InetAddress.getByName(member.getHost());
// Create a StringBuffer for the JSON message.
// Create a StringBuffer for the JSON message. JSONArray jsonArray = new JSONArray();
JSONArray jsonArray = new JSONArray(); GossipService.LOGGER.debug("Sending memberlist to " + dest + ":" + member.getPort());
GossipService.LOGGER.debug("Sending memberlist to " + dest + ":" + member.getPort()); GossipService.LOGGER.debug("---------------------");
GossipService.LOGGER.debug("---------------------");
// First write myself, append the JSON representation of the member to the buffer.
// First write myself, append the JSON representation of the member to the buffer. jsonArray.put(me.toJSONObject());
jsonArray.put(me.toJSONObject()); GossipService.LOGGER.debug(me);
GossipService.LOGGER.debug(me);
// Then write the others.
// Then write the others. for (int i = 0; i < memberList.size(); i++) {
for (int i=0; i<memberList.size(); i++) { LocalGossipMember other = memberList.get(i);
LocalGossipMember other = memberList.get(i); // Append the JSON representation of the member to the buffer.
// Append the JSON representation of the member to the buffer. jsonArray.put(other.toJSONObject());
jsonArray.put(other.toJSONObject()); GossipService.LOGGER.debug(other);
GossipService.LOGGER.debug(other); }
} GossipService.LOGGER.debug("---------------------");
GossipService.LOGGER.debug("---------------------");
// Write the objects to a byte array.
// Write the objects to a byte array. byte[] json_bytes = jsonArray.toString().getBytes();
byte[] json_bytes = jsonArray.toString().getBytes();
int packet_length = json_bytes.length;
int packet_length = json_bytes.length;
if (packet_length < GossipManager.MAX_PACKET_SIZE) {
if (packet_length < GossipManager.MAX_PACKET_SIZE) {
// Convert the packet length to the byte representation of the int.
// Convert the packet length to the byte representation of the int. byte[] length_bytes = new byte[4];
byte[] length_bytes = new byte[4]; length_bytes[0] = (byte) (packet_length >> 24);
length_bytes[0] =(byte)( packet_length >> 24 ); length_bytes[1] = (byte) ((packet_length << 8) >> 24);
length_bytes[1] =(byte)( (packet_length << 8) >> 24 ); length_bytes[2] = (byte) ((packet_length << 16) >> 24);
length_bytes[2] =(byte)( (packet_length << 16) >> 24 ); length_bytes[3] = (byte) ((packet_length << 24) >> 24);
length_bytes[3] =(byte)( (packet_length << 24) >> 24 );
GossipService.LOGGER.debug("Sending message (" + packet_length + " bytes): "
+ jsonArray.toString());
GossipService.LOGGER.debug("Sending message ("+packet_length+" bytes): " + jsonArray.toString());
ByteBuffer byteBuffer = ByteBuffer.allocate(4 + json_bytes.length);
ByteBuffer byteBuffer = ByteBuffer.allocate(4 + json_bytes.length); byteBuffer.put(length_bytes);
byteBuffer.put(length_bytes); byteBuffer.put(json_bytes);
byteBuffer.put(json_bytes); byte[] buf = byteBuffer.array();
byte[] buf = byteBuffer.array();
DatagramSocket socket = new DatagramSocket();
DatagramSocket socket = new DatagramSocket(); DatagramPacket datagramPacket = new DatagramPacket(buf, buf.length, dest,
DatagramPacket datagramPacket = new DatagramPacket(buf, buf.length, dest, member.getPort()); member.getPort());
socket.send(datagramPacket); socket.send(datagramPacket);
socket.close(); socket.close();
} else { } else {
GossipService.LOGGER.error("The length of the to be send message is too large (" + packet_length + " > " + GossipManager.MAX_PACKET_SIZE + ")."); GossipService.LOGGER.error("The length of the to be send message is too large ("
} + packet_length + " > " + GossipManager.MAX_PACKET_SIZE + ").");
} }
}
} catch (IOException e1) {
e1.printStackTrace(); } catch (IOException e1) {
} e1.printStackTrace();
} }
} }
} }
}

76
src/main/java/com/google/code/gossip/manager/random/RandomActiveGossipThread.java
View File

@ -1,38 +1,38 @@
package com.google.code.gossip.manager.random; package com.google.code.gossip.manager.random;
import java.util.ArrayList; import java.util.ArrayList;
import java.util.Random; import java.util.Random;
import com.google.code.gossip.GossipService; import com.google.code.gossip.GossipService;
import com.google.code.gossip.LocalGossipMember; import com.google.code.gossip.LocalGossipMember;
import com.google.code.gossip.manager.GossipManager; import com.google.code.gossip.manager.GossipManager;
import com.google.code.gossip.manager.impl.SendMembersActiveGossipThread; import com.google.code.gossip.manager.impl.SendMembersActiveGossipThread;
public class RandomActiveGossipThread extends SendMembersActiveGossipThread { public class RandomActiveGossipThread extends SendMembersActiveGossipThread {
/** The Random used for choosing a member to gossip with. */ /** The Random used for choosing a member to gossip with. */
private Random _random; private Random _random;
public RandomActiveGossipThread(GossipManager gossipManager) { public RandomActiveGossipThread(GossipManager gossipManager) {
super(gossipManager); super(gossipManager);
_random = new Random(); _random = new Random();
} }
/** /**
* [The selectToSend() function.] * [The selectToSend() function.] Find a random peer from the local membership list. In the case
* Find a random peer from the local membership list. * where this client is the only member in the list, this method will return null.
* In the case where this client is the only member in the list, this method will return null. *
* @return Member random member if list is greater than 1, null otherwise * @return Member random member if list is greater than 1, null otherwise
*/ */
protected LocalGossipMember selectPartner(ArrayList<LocalGossipMember> memberList) { protected LocalGossipMember selectPartner(ArrayList<LocalGossipMember> memberList) {
LocalGossipMember member = null; LocalGossipMember member = null;
if (memberList.size() > 0) { if (memberList.size() > 0) {
int randomNeighborIndex = _random.nextInt(memberList.size()); int randomNeighborIndex = _random.nextInt(memberList.size());
member = memberList.get(randomNeighborIndex); member = memberList.get(randomNeighborIndex);
} else { } else {
GossipService.LOGGER.debug("I am alone in this world."); GossipService.LOGGER.debug("I am alone in this world.");
} }
return member; return member;
} }
} }

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

@ -1,14 +1,16 @@
package com.google.code.gossip.manager.random; package com.google.code.gossip.manager.random;
import java.util.ArrayList; import java.util.ArrayList;
import com.google.code.gossip.GossipMember; import com.google.code.gossip.GossipMember;
import com.google.code.gossip.GossipSettings; import com.google.code.gossip.GossipSettings;
import com.google.code.gossip.manager.GossipManager; import com.google.code.gossip.manager.GossipManager;
import com.google.code.gossip.manager.impl.OnlyProcessReceivedPassiveGossipThread; import com.google.code.gossip.manager.impl.OnlyProcessReceivedPassiveGossipThread;
public class RandomGossipManager extends GossipManager { public class RandomGossipManager extends GossipManager {
public RandomGossipManager(String address, int port, String id, GossipSettings settings, ArrayList<GossipMember> gossipMembers) { public RandomGossipManager(String address, int port, String id, GossipSettings settings,
super(OnlyProcessReceivedPassiveGossipThread.class, RandomActiveGossipThread.class, address, port, id, settings, gossipMembers); ArrayList<GossipMember> gossipMembers) {
} super(OnlyProcessReceivedPassiveGossipThread.class, RandomActiveGossipThread.class, address,
} port, id, settings, gossipMembers);
}
}