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

GOSSIP-25 Create reaper process to expire per-node data

Browse Source
This commit is contained in:
Edward Capriolo
2016-10-07 02:03:43 -04:00
parent daea6edb15
commit f35dddd8f2
12 changed files with 206 additions and 46 deletions
Download Patch File Download Diff File Expand all files Collapse all files

31
src/main/java/org/apache/gossip/manager/GossipManager.java
View File

@ -22,6 +22,7 @@ import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map.Entry;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentSkipListMap;
import java.util.concurrent.ExecutorService;
@ -67,14 +68,20 @@ public abstract class GossipManager implements NotificationListener {
private ExecutorService gossipThreadExecutor;
private GossipCore gossipCore;
private final GossipCore gossipCore;
private final DataReaper dataReaper;
private final Clock clock;
public GossipManager(String cluster,
URI uri, String id, GossipSettings settings,
List<GossipMember> gossipMembers, GossipListener listener) {
this.settings = settings;
this.gossipCore = new GossipCore(this);
gossipCore = new GossipCore(this);
clock = new SystemClock();
dataReaper = new DataReaper(gossipCore, clock);
me = new LocalGossipMember(cluster, uri, id, System.currentTimeMillis(), this,
settings.getCleanupInterval());
members = new ConcurrentSkipListMap<>();
@ -192,6 +199,7 @@ public abstract class GossipManager implements NotificationListener {
gossipThreadExecutor.execute(passiveGossipThread);
activeGossipThread = new ActiveGossipThread(this, this.gossipCore);
activeGossipThread.init();
dataReaper.init();
GossipService.LOGGER.debug("The GossipService is started.");
}
@ -202,6 +210,7 @@ public abstract class GossipManager implements NotificationListener {
gossipServiceRunning.set(false);
gossipThreadExecutor.shutdown();
gossipCore.shutdown();
dataReaper.close();
if (passiveGossipThread != null) {
passiveGossipThread.shutdown();
}
@ -218,7 +227,10 @@ public abstract class GossipManager implements NotificationListener {
}
}
public void gossipData(GossipDataMessage message){
public void gossipPerNodeData(GossipDataMessage message){
Objects.nonNull(message.getKey());
Objects.nonNull(message.getTimestamp());
Objects.nonNull(message.getPayload());
message.setNodeId(me.getId());
gossipCore.addPerNodeData(message);
}
@ -228,8 +240,19 @@ public abstract class GossipManager implements NotificationListener {
if (j == null){
return null;
} else {
return j.get(key);
GossipDataMessage l = j.get(key);
if (l == null){
return null;
}
if (l.getExpireAt() != null && l.getExpireAt() < clock.currentTimeMillis()) {
return null;
}
return l;
}
}
public DataReaper getDataReaper() {
return dataReaper;
}
}