server

Documentation

Index

• type AppendEntries
• type ApplyRequest
• type OptFunc
  • func WithGetStateFunc(g getStateFunc) OptFunc
  • func WithLogger(logger *slog.Logger) OptFunc
  • func WithTimeout(timeout int) OptFunc
• type RPCResponse
• type RPCServer
  • func New(id int, port uint16, opts ...OptFunc) (*RPCServer, error)
  • func (s *RPCServer) AppendEntries(ctx context.Context, in *pb.AppendEntriesRequest) (*pb.AppendEntriesResponse, error)
  • func (s *RPCServer) ApplyEntry(ctx context.Context, in *pb.ApplyRequest) (*pb.ApplyResponse, error)
  • func (s *RPCServer) GetPeerConn(index uint) *grpc.ClientConn
  • func (s *RPCServer) Notify(state bool)
  • func (s *RPCServer) RequestVote(ctx context.Context, in *pb.VoteRequest) (*pb.VoteResponse, error)
  • func (s *RPCServer) Run(ctx context.Context, peers map[uint]string, secure bool) error
  • func (s *RPCServer) SendAppendEntries(ctx context.Context, node uint, req AppendEntries) (*RPCResponse, error)
  • func (s *RPCServer) SendRequestVote(ctx context.Context, node uint, req VoteRequest) (*RPCResponse, error)
  • func (s *RPCServer) SetAppendEntryRPCChan(appendEntriesChan chan AppendEntries) Server
  • func (s *RPCServer) SetApplyEntryRPCChan(applyEntryChan chan ApplyRequest) Server
  • func (s *RPCServer) SetVoteRPCChan(voteChan chan VoteRequest) Server
  • func (s *RPCServer) Stop()
• type Sender
• type Server
• type VoteRequest

Types

type AppendEntries

type AppendEntries struct {
	// Term is the observed term of the leader.
	Term uint64
	// LeaderId is the ID of the leader.
	LeaderId int32
	// PrevLogIndex is the index of the log entry immediately preceding the new ones.
	PrevLogIndex uint64
	// PrevLogTerm is the term of the log entry immediately preceding the new ones.
	PrevLogTerm uint64
	// Entries are the log entries to append.
	Entries log.LogEntries
	// LeaderCommit is the leader's commit index.
	LeaderCommit uint64
	// ResponseChan is the channel to send the response to.
	ResponseChan chan RPCResponse
}

AppendEntries is a message sent to the a raft node to append entries to the log.

type ApplyRequest

type ApplyRequest struct {
	// Sn is the serial number of the command.
	Sn int64
	// Command is the command to apply.
	Command []byte
	// ResponseChan is the channel to send the response to.
	ResponseChan chan RPCResponse
}

ApplyRequest is a message sent to the raft node to apply a command.

type OptFunc

type OptFunc func(o *options)

func WithGetStateFunc

func WithGetStateFunc(g getStateFunc) OptFunc

WithGetCStateFunc sets the function to get the current term.

func WithLogger

func WithLogger(logger *slog.Logger) OptFunc

WithLogger sets the logger.

func WithTimeout

func WithTimeout(timeout int) OptFunc

WithTimeout sets the timeout.

type RPCResponse

type RPCResponse struct {
	// Term is the current term of the node.
	Term uint64
	// Response is the response to the request. It is true if the request was
	// accepted.
	Response bool
}

RPCResponse is a response to an RPC request.

type RPCServer

type RPCServer struct {
	pb.UnimplementedAppendEntriesServer
	pb.UnimplementedVoteServer
	pb.UnimplementedApplyEntryServer
	// contains filtered or unexported fields
}

RPCServer is a RPC server that handles all the RPC communications. It implements the Server interface and uses the grpc protocol.

func New

func New(id int, port uint16, opts ...OptFunc) (*RPCServer, error)

New returns a new RPCServer.

func (*RPCServer) AppendEntries

func (s *RPCServer) AppendEntries(ctx context.Context, in *pb.AppendEntriesRequest) (*pb.AppendEntriesResponse, error)

AppendEntries is called by leaders to replicate log entries. It receives an append entries request and sends a response. If returns early if the request term is less than the current term.

func (*RPCServer) ApplyEntry

func (s *RPCServer) ApplyEntry(ctx context.Context, in *pb.ApplyRequest) (*pb.ApplyResponse, error)

ApplyEntry is called by clients to apply a command. It receives an apply entry request and sends a response. It blocks until the command is committed. If the node is not the leader, it returns an error.

func (*RPCServer) GetPeerConn

func (s *RPCServer) GetPeerConn(index uint) *grpc.ClientConn

GetPeerConn returns a grpc connection to be used to send RPCs to the peer.

func (*RPCServer) Notify

func (s *RPCServer) Notify(state bool)

Notify is used to observe the state of the node. It implements the Observer interface.

func (*RPCServer) RequestVote

func (s *RPCServer) RequestVote(ctx context.Context, in *pb.VoteRequest) (*pb.VoteResponse, error)

RequestVote is called by candidates to gather votes. It receives a vote request and sends a response. If returns early if the request term is less than the current term.

func (*RPCServer) Run

func (s *RPCServer) Run(ctx context.Context, peers map[uint]string, secure bool) error

Run starts the server.

func (*RPCServer) SendAppendEntries

func (s *RPCServer) SendAppendEntries(ctx context.Context, node uint, req AppendEntries) (*RPCResponse, error)

SendAppendEntries sends an append entries to a node. The node is identified by its ID. It returns an error if the node is not connected and marks the node as dead.

func (*RPCServer) SendRequestVote

func (s *RPCServer) SendRequestVote(ctx context.Context, node uint, req VoteRequest) (*RPCResponse, error)

SendRequestVote sends a request vote to a node. The node is identified by its ID. It returns an error if the node is not connected and marks the node as dead.

func (*RPCServer) SetAppendEntryRPCChan

func (s *RPCServer) SetAppendEntryRPCChan(appendEntriesChan chan AppendEntries) Server

SetAppendEntryRPCChan sets the channel to send entries to replicate to the fsm.

func (*RPCServer) SetApplyEntryRPCChan

func (s *RPCServer) SetApplyEntryRPCChan(applyEntryChan chan ApplyRequest) Server

SetApplyEntryRPCChan sets the channel to send entries to apply to the fsm.

func (*RPCServer) SetVoteRPCChan

func (s *RPCServer) SetVoteRPCChan(voteChan chan VoteRequest) Server

SetVoteRPCChan sets the channel to send vote requests to the fsm.

func (*RPCServer) Stop

func (s *RPCServer) Stop()

Stop stops the server.

type Sender

type Sender interface {
	// SendAppendEntries sends an AppendEntries RPC to the given node.
	SendAppendEntries(ctx context.Context, node uint, req AppendEntries) (*RPCResponse, error)
	// SendRequestVote sends a VoteRequest RPC to the given node.
	SendRequestVote(ctx context.Context, node uint, req VoteRequest) (*RPCResponse, error)
}

Sender is the interface that wraps the basic methods for sending RPCs to other raft nodes.

type Server

type Server interface {
	Sender
	// Run starts the server.
	Run(ctx context.Context, peers map[uint]string, testMode bool) error
	// SetVoteRPCChan sets the channel to send vote requests to the fsm.
	SetVoteRPCChan(voteChan chan VoteRequest) Server
	// SetAppendEntryRPCChan sets the channel to send entries to replicate to the fsm.
	SetAppendEntryRPCChan(appendEntriesChan chan AppendEntries) Server
	// SetApplyEntryRPCChan sets the channel to send entries to apply to the fsm.
	SetApplyEntryRPCChan(applyEntryChan chan ApplyRequest) Server
	// Stop stops the server.
	Stop()
}

Server is the interface that wraps the basic methods for communicating with with a raft node.

type VoteRequest

type VoteRequest struct {
	// Term is the candidate's term.
	Term uint64
	// CandidateId is the candidate requesting the vote.
	CandidateId int32
	// LastLogIndex is the index of the candidate's last log entry.
	LastLogIndex uint64
	// LastLogTerm is the term of the candidate's last log entry.
	LastLogTerm uint64
	// ResponseChan is the channel to send the response to.
	ResponseChan chan RPCResponse
}

VoteRequest is a message sent to the raft node to request a vote.