remove RaftNode construct code to init

raft-java-core/src/main/java/com/github/wenweihu86/raft/PeerId.java

@@ -0,0 +1,13 @@
+package com.github.wenweihu86.raft;
+
+public class PeerId {
+    private Integer peerId;
+
+    public PeerId(Integer peerId) {
+        this.peerId = peerId;
+    }
+
+    public Integer getPeerId() {
+        return peerId;
+    }
+}

raft-java-core/src/main/java/com/github/wenweihu86/raft/RaftNode.java

@@ -17,6 +17,7 @@
 import java.io.IOException;
 import java.util.*;
 import java.util.concurrent.*;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.locks.*;
 
 /**
@@ -40,13 +41,13 @@ public enum NodeState {
 
     private RaftOptions raftOptions;
     private RaftMessage.Configuration configuration;
-    private ConcurrentMap<Integer, Peer> peerMap = new ConcurrentHashMap<>();
+    private ConcurrentMap<PeerId, Peer> peerMap = new ConcurrentHashMap<>();
     private RaftMessage.Server localServer;
     private StateMachine stateMachine;
     private SegmentedLog raftLog;
     private Snapshot snapshot;
 
-    private NodeState state = NodeState.STATE_FOLLOWER;
+    private volatile NodeState state;
     // 服务器最后一次知道的任期号（初始化为 0，持续递增）
     private long currentTerm;
     // 在当前获得选票的候选人的Id
@@ -66,6 +67,8 @@ public enum NodeState {
     private ScheduledFuture electionScheduledFuture;
     private ScheduledFuture heartbeatScheduledFuture;
 
+    private AtomicBoolean hasInit = new AtomicBoolean(false);
+
     public RaftNode(RaftOptions raftOptions,
                     List<RaftMessage.Server> servers,
                     RaftMessage.Server localServer,
@@ -79,14 +82,23 @@ public RaftNode(RaftOptions raftOptions,
 
         this.localServer = localServer;
         this.stateMachine = stateMachine;
+    }
+
+    public void init() {
+
+        if (!hasInit.compareAndSet(false, true)) {
+            throw new IllegalStateException("Raft Node has init before");
+        }
 
         // load log and snapshot
         raftLog = new SegmentedLog(raftOptions.getDataDir(), raftOptions.getMaxSegmentFileSize());
         snapshot = new Snapshot(raftOptions.getDataDir());
         snapshot.reload();
 
         currentTerm = raftLog.getMetaData().getCurrentTerm();
+
         votedFor = raftLog.getMetaData().getVotedFor();
+
         commitIndex = Math.max(snapshot.getMetaData().getLastIncludedIndex(), commitIndex);
         // discard old log entries
         if (snapshot.getMetaData().getLastIncludedIndex() > 0
@@ -110,15 +122,16 @@ public RaftNode(RaftOptions raftOptions,
             }
         }
         lastAppliedIndex = commitIndex;
-    }
 
-    public void init() {
+        // init state is FOLLOWER
+        state = NodeState.STATE_FOLLOWER;
+
         for (RaftMessage.Server server : configuration.getServersList()) {
             if (!peerMap.containsKey(server.getServerId())
                     && server.getServerId() != localServer.getServerId()) {
                 Peer peer = new Peer(server);
                 peer.setNextIndex(raftLog.getLastLogIndex() + 1);
-                peerMap.put(server.getServerId(), peer);
+                peerMap.put(new PeerId(server.getServerId()), peer);
             }
         }
 
@@ -408,7 +421,7 @@ public void applyConfiguration(RaftMessage.LogEntry entry) {
                         && server.getServerId() != localServer.getServerId()) {
                     Peer peer = new Peer(server);
                     peer.setNextIndex(raftLog.getLastLogIndex() + 1);
-                    peerMap.put(server.getServerId(), peer);
+                    peerMap.put(new PeerId(server.getServerId()), peer);
                 }
             }
             LOG.info("new conf is {}, leaderId={}", PRINTER.print(newConfiguration), leaderId);
@@ -928,6 +941,18 @@ private RaftMessage.InstallSnapshotRequest buildInstallSnapshotRequest(
         return requestBuilder.build();
     }
 
+    public void removePeer(PeerId peerId) {
+        peerMap.remove(peerId);
+    }
+
+    public boolean containsPeer(PeerId peerId) {
+        return peerMap.containsKey(peerId);
+    }
+
+    public void addPeer(Peer peer) {
+        peerMap.put(new PeerId(peer.getServer().getServerId()), peer);
+    }
+
     public Lock getLock() {
         return lock;
     }
@@ -992,14 +1017,6 @@ public RaftMessage.Server getLocalServer() {
         return localServer;
     }
 
-    public NodeState getState() {
-        return state;
-    }
-
-    public ConcurrentMap<Integer, Peer> getPeerMap() {
-        return peerMap;
-    }
-
     public ExecutorService getExecutorService() {
         return executorService;
     }

raft-java-core/src/main/java/com/github/wenweihu86/raft/service/impl/RaftClientServiceImpl.java

@@ -1,6 +1,7 @@
 package com.github.wenweihu86.raft.service.impl;
 
 import com.github.wenweihu86.raft.Peer;
+import com.github.wenweihu86.raft.PeerId;
 import com.github.wenweihu86.raft.RaftNode;
 import com.github.wenweihu86.raft.proto.RaftMessage;
 import com.github.wenweihu86.raft.service.RaftClientService;
@@ -99,7 +100,7 @@ public RaftMessage.AddPeersResponse addPeers(RaftMessage.AddPeersRequest request
             return responseBuilder.build();
         }
         for (RaftMessage.Server server : request.getServersList()) {
-            if (raftNode.getPeerMap().containsKey(server.getServerId())) {
+            if (raftNode.containsPeer(new PeerId(server.getServerId()))) {
                 LOG.warn("already be added/adding to configuration");
                 responseBuilder.setResMsg("already be added/adding to configuration");
                 return responseBuilder.build();
@@ -110,7 +111,7 @@ public RaftMessage.AddPeersResponse addPeers(RaftMessage.AddPeersRequest request
             final Peer peer = new Peer(server);
             peer.setNextIndex(1);
             requestPeers.add(peer);
-            raftNode.getPeerMap().putIfAbsent(server.getServerId(), peer);
+            raftNode.addPeer(peer);
             raftNode.getExecutorService().submit(new Runnable() {
                 @Override
                 public void run() {
@@ -163,7 +164,7 @@ public void run() {
             try {
                 for (Peer peer : requestPeers) {
                     peer.getRpcClient().stop();
-                    raftNode.getPeerMap().remove(peer.getServer().getServerId());
+                    raftNode.removePeer(new PeerId(peer.getServer().getServerId()));
                 }
             } finally {
                 raftNode.getLock().unlock();