Implement Raft leader heartbeats (eea78512-6817-44f0-849c-52f5003b0111)

1 files changed, 29 insertions, 0 deletions

diff --git a/src/main/java/protocols/implementations/VSRaftProtocol.java b/src/main/java/protocols/implementations/VSRaftProtocol.java
index 8b919f6..56e758d 100644
--- a/src/main/java/protocols/implementations/VSRaftProtocol.java
+++ b/src/main/java/protocols/implementations/VSRaftProtocol.java
@@ -76,6 +76,7 @@ public class VSRaftProtocol extends VSAbstractProtocol {
      * @see protocols.VSAbstractProtocol#onServerStart()
      */
     public void onServerStart() {
+        becomeLeader();
     }
 
     /* (non-Javadoc)
@@ -94,6 +95,9 @@ public class VSRaftProtocol extends VSAbstractProtocol {
      * @see protocols.VSAbstractProtocol#onServerSchedule()
      */
     public void onServerSchedule() {
+        if (isLeader) {
+            sendHeartbeat();
+        }
     }
 
     /* (non-Javadoc)
@@ -136,4 +140,29 @@ public class VSRaftProtocol extends VSAbstractProtocol {
             ackPids.clear();
         }
     }
+
+    /**
+     * Transitions this process into the leader role and starts heartbeats.
+     */
+    private void becomeLeader() {
+        isLeader = true;
+        isCandidate = false;
+        leaderId = process.getProcessID();
+        lastHeartbeatTime = process.getTime();
+        sendHeartbeat();
+    }
+
+    /**
+     * Sends a heartbeat and schedules the next leader heartbeat interval.
+     */
+    private void sendHeartbeat() {
+        VSMessage heartbeat = new VSMessage();
+        heartbeat.setString("type", "heartbeat");
+        heartbeat.setInteger("term", currentTerm);
+        heartbeat.setInteger("leaderId", leaderId);
+        sendMessage(heartbeat);
+
+        lastHeartbeatTime = process.getTime();
+        scheduleAt(process.getTime() + getLong("heartbeatInterval"));
+    }
 }