fix: some error

Author: treeforest

config.go

@@ -1,12 +1,19 @@
 package raft
 
 import (
+	"github.com/bwmarrin/snowflake"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials/insecure"
 	"time"
 )
 
 type Config struct {
+	// MemberId 节点的id
+	MemberId uint64
+
+	// Address 节点的监听地址
+	Address string
+
 	// MaxLogEntriesPerRequest 每次最多请求的日志条目
 	MaxLogEntriesPerRequest uint64
 
@@ -16,9 +23,6 @@ type Config struct {
 	// ElectionTimeout 选举超时
 	ElectionTimeout time.Duration
 
-	// ElectionTimeoutThresholdPercent 选举超时的阈值
-	ElectionTimeoutThresholdPercent float64
-
 	// DialTimeout 拨号超时时间
 	DialTimeout time.Duration
 
@@ -28,20 +32,29 @@ type Config struct {
 	// ServerOptions 服务端参数设置
 	ServerOptions []grpc.ServerOption
 
+	// SnapshotPath 快照存储路径
 	SnapshotPath string
+
+	// LogPath 日志文件路径
+	LogPath string
 }
 
 func DefaultConfig() *Config {
+	node, _ := snowflake.NewNode(time.Now().UnixNano())
 	return &Config{
-		HeartbeatInterval: time.Millisecond * 100,
-		ElectionTimeout:   time.Millisecond * 150,
-		DialTimeout:       time.Second,
+		MemberId:                uint64(node.Generate().Int64()),
+		Address:                 "localhost:4399",
+		MaxLogEntriesPerRequest: 40,
+		HeartbeatInterval:       time.Millisecond * 100,
+		ElectionTimeout:         time.Millisecond * 150,
+		DialTimeout:             time.Second,
 		DialOptions: []grpc.DialOption{
 			grpc.WithTransportCredentials(insecure.NewCredentials()),
 		},
 		ServerOptions: []grpc.ServerOption{
 			grpc.Creds(insecure.NewCredentials()),
 		},
 		SnapshotPath: "./snapshot",
+		LogPath:      "./log",
 	}
 }

event.go

@@ -0,0 +1,117 @@
+package main
+
+import (
+	"encoding/json"
+	"flag"
+	"fmt"
+	"github.com/gin-gonic/gin"
+	log "github.com/treeforest/logger"
+	"github.com/treeforest/raft"
+	"os"
+	"os/signal"
+	"strings"
+	"sync"
+	"time"
+)
+
+type Command struct {
+	Key   string
+	Value string
+}
+
+type Server struct {
+	raft.StateMachine
+	peer   raft.Raft
+	r      *gin.Engine
+	state  map[string]string
+	locker sync.RWMutex
+	pool   *sync.Pool
+}
+
+func New() *Server {
+	return &Server{
+		state:  map[string]string{},
+		locker: sync.RWMutex{},
+		pool: &sync.Pool{
+			New: func() interface{} {
+				return new(Command)
+			},
+		},
+	}
+}
+
+// Save 读取状态机快照
+func (s *Server) Save() ([]byte, error) {
+	s.locker.RLock()
+	defer s.locker.RUnlock()
+	return json.Marshal(s.state)
+}
+
+// Recovery 从快照中恢复状态机状态
+func (s *Server) Recovery(state []byte) error {
+	s.locker.Lock()
+	defer s.locker.Unlock()
+	return json.Unmarshal(state, &s.state)
+}
+
+// Apply 状态机执行命令的回调函数
+func (s *Server) Apply(commandName string, command []byte) {
+	c := s.pool.Get().(*Command)
+	_ = json.Unmarshal(command, c)
+	switch commandName {
+	case "set":
+		s.state[c.Key] = c.Value
+	case "del":
+		delete(s.state, c.Key)
+	}
+	s.pool.Put(c)
+}
+
+func (s *Server) Serve(addr string) {
+	r := gin.Default()
+	r.POST("/set", func(c *gin.Context) {
+		if !s.peer.IsLeader() {
+			c.JSON(200, gin.H{"code": -1, "leader": s.peer.LeaderAddress()})
+		}
+		key := c.Query("key")
+		val := c.Query("val")
+
+	})
+
+	s.r = r
+	if err := s.r.Run(addr); err != nil {
+		log.Error(err)
+	}
+}
+
+func main() {
+	port := flag.Int("port", 0, "raft server port")
+	addr := flag.String("addr", "localhost:0", "web address")
+	existing := flag.String("existing", "", "existing raft member")
+	flag.Parse()
+
+	s := New()
+
+	config := raft.DefaultConfig()
+	config.Address = fmt.Sprintf("localhost:%d", *port)
+	config.LogPath = fmt.Sprintf("%d", *port)
+
+	peer := raft.New(raft.DefaultConfig(), s)
+	if err := peer.Start(); err != nil {
+		log.Fatal(err)
+	}
+
+	peer.Join(strings.Split(*existing, ","))
+
+	go func() {
+		s.peer = peer
+		s.Serve(*addr)
+	}()
+
+	done := make(chan os.Signal, 1)
+	signal.Notify(done, os.Interrupt, os.Kill)
+	<-done
+
+	srv.Stop()
+	time.Sleep(time.Millisecond * 500)
+}

event_dispatcher.go

@@ -3,25 +3,38 @@ module github.com/treeforest/raft
 go 1.17
 
 require (
+	github.com/bwmarrin/snowflake v0.3.0
+	github.com/gin-gonic/gin v1.7.7
 	github.com/gogo/protobuf v1.3.2
 	github.com/golang/protobuf v1.5.2
-	github.com/stretchr/testify v1.7.0
 	github.com/syndtr/goleveldb v1.0.0
 	github.com/treeforest/logger v0.1.0
 	google.golang.org/grpc v1.45.0
 )
 
 require (
-	github.com/davecgh/go-spew v1.1.0 // indirect
+	github.com/davecgh/go-spew v1.1.1 // indirect
 	github.com/fatih/color v1.13.0 // indirect
+	github.com/gin-contrib/sse v0.1.0 // indirect
+	github.com/go-playground/locales v0.13.0 // indirect
+	github.com/go-playground/universal-translator v0.17.0 // indirect
+	github.com/go-playground/validator/v10 v10.4.1 // indirect
 	github.com/golang/snappy v0.0.0-20180518054509-2e65f85255db // indirect
+	github.com/json-iterator/go v1.1.9 // indirect
+	github.com/leodido/go-urn v1.2.0 // indirect
 	github.com/mattn/go-colorable v0.1.9 // indirect
 	github.com/mattn/go-isatty v0.0.14 // indirect
+	github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421 // indirect
+	github.com/modern-go/reflect2 v0.0.0-20180701023420-4b7aa43c6742 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
+	github.com/stretchr/testify v1.7.0 // indirect
+	github.com/ugorji/go/codec v1.1.7 // indirect
+	golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9 // indirect
 	golang.org/x/net v0.0.0-20201021035429-f5854403a974 // indirect
 	golang.org/x/sys v0.0.0-20210630005230-0f9fa26af87c // indirect
 	golang.org/x/text v0.3.3 // indirect
 	google.golang.org/genproto v0.0.0-20200526211855-cb27e3aa2013 // indirect
 	google.golang.org/protobuf v1.26.0 // indirect
+	gopkg.in/yaml.v2 v2.2.8 // indirect
 	gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c // indirect
 )