rqd

Documentation

Index

• Constants
• Variables
• func GetLogPackHeader(r io.Reader) uint32
• func LogPackHeader(typ uint32) []byte
• func NewFSMHandlers(s store.Store) map[serverpb.RaftLogCommand]FSMHandleFunc
• func NewTLSTransport(addr string, advertise net.Addr, maxPool int, timeout time.Duration, ...) (*raft.NetworkTransport, error)
• func NewTLSTransportWithGenerator(addr string, advertise net.Addr, config *raft.NetworkTransportConfig) (*raft.NetworkTransport, error)
• func RaftLogPayloadKey(m *serverpb.RaftLogPayload) uint64
• func UnpackLog(r io.Reader) ([]*serverpb.RaftLogPayload, error)
• type ApplyFunc
• type FSM
  • func (f *FSM) Apply(log *raft.Log) any
  • func (f *FSM) Restore(rc io.ReadCloser) error
  • func (f *FSM) Snapshot() (raft.FSMSnapshot, error)
• type FSMHandleFunc
• type FSMHandlers
  • func (h *FSMHandlers) Del(payload *serverpb.RaftLogPayload) error
  • func (h *FSMHandlers) Set(payload *serverpb.RaftLogPayload) error
  • func (h *FSMHandlers) SetWithTTL(payload *serverpb.RaftLogPayload) error
  • func (h *FSMHandlers) TrySet(payload *serverpb.RaftLogPayload) error
  • func (h *FSMHandlers) TrySetWithTTL(payload *serverpb.RaftLogPayload) error
• type Raft
  • func NewRaftWithOptions(opts ...RaftServerOption) (*Raft, error)
  • func (r *Raft) AddCluster(id raft.ServerID, addr raft.ServerAddress) error
  • func (r *Raft) Term() uint64
• type RaftApply
  • func NewRaftMultipleLogApply(ctx context.Context, maxLimit int32, deadline, applyTimeout time.Duration, ...) RaftApply
  • func NewRaftSingeLogApply(fc ApplyFunc) RaftApply
• type RaftServerOption
  • func RaftWithBoltLogStore(path string) RaftServerOption
  • func RaftWithBootstrap() RaftServerOption
  • func RaftWithClusters(clusters []raft.Server) RaftServerOption
  • func RaftWithConfig(cfg *raft.Config) RaftServerOption
  • func RaftWithContext(ctx context.Context) RaftServerOption
  • func RaftWithEmpty() RaftServerOption
  • func RaftWithFileSnapshotStore(path string, retain int, logOut io.Writer) RaftServerOption
  • func RaftWithStdFSM(store store.Store) RaftServerOption
  • func RaftWithStdStableStore(store store.Store) RaftServerOption
  • func RaftWithTCPTransport(addr string, maxPool int, timeout time.Duration, logOut io.Writer) RaftServerOption
• type RaftTLStreamLayer
  • func (r RaftTLStreamLayer) Accept() (net.Conn, error)
  • func (r RaftTLStreamLayer) Addr() net.Addr
  • func (r RaftTLStreamLayer) Close() error
  • func (r RaftTLStreamLayer) Dial(addr raft.ServerAddress, timeout time.Duration) (net.Conn, error)
• type Server
  • func NewServer(cfg *config.Config) (*Server, error)
  • func (s *Server) ListenAndServe() error
  • func (s *Server) Shutdown()
• type Snapshot
  • func (s *Snapshot) Persist(sink raft.SnapshotSink) error
  • func (s *Snapshot) Release()
• type StableStore
  • func NewStableStore(s store.Store) (*StableStore, error)
  • func (s *StableStore) Get(key []byte) ([]byte, error)
  • func (s *StableStore) GetUint64(key []byte) (uint64, error)
  • func (s *StableStore) Set(key []byte, val []byte) error
  • func (s *StableStore) SetUint64(key []byte, val uint64) error

Constants

const (
	SingleLogPack uint32 = iota
	MultipleLogPack
)

const (
	StoreSuffix = "data"
	RaftLog     = "raft-log.db"
)

Variables

var (
	ErrApplyLogTimeTravelDone = errors.New("raft apply log time-travel done")
	ErrApplyLogDone           = errors.New("raft apply log done")
)

Functions

func GetLogPackHeader

func GetLogPackHeader(r io.Reader) uint32

func LogPackHeader

func LogPackHeader(typ uint32) []byte

func NewFSMHandlers

func NewFSMHandlers(s store.Store) map[serverpb.RaftLogCommand]FSMHandleFunc

func NewTLSTransport

func NewTLSTransport(
	addr string,
	advertise net.Addr,
	maxPool int,
	timeout time.Duration,
	logOutput io.Writer,
	cfg *tls.Config,
) (*raft.NetworkTransport, error)

func NewTLSTransportWithGenerator

func NewTLSTransportWithGenerator(
	addr string,
	advertise net.Addr,
	config *raft.NetworkTransportConfig,
) (*raft.NetworkTransport, error)

func RaftLogPayloadKey

func RaftLogPayloadKey(m *serverpb.RaftLogPayload) uint64

func UnpackLog

func UnpackLog(r io.Reader) ([]*serverpb.RaftLogPayload, error)

Types

type ApplyFunc

type ApplyFunc func(cmd []byte, timeout time.Duration) raft.ApplyFuture

type FSM

type FSM struct {
	Term     *uint64
	Handlers map[serverpb.RaftLogCommand]FSMHandleFunc
	Store    store.Store
}

func (*FSM) Apply

func (f *FSM) Apply(log *raft.Log) any

func (*FSM) Restore

func (f *FSM) Restore(rc io.ReadCloser) error

func (*FSM) Snapshot

func (f *FSM) Snapshot() (raft.FSMSnapshot, error)

type FSMHandleFunc

type FSMHandleFunc func(*serverpb.RaftLogPayload) error

type FSMHandlers

type FSMHandlers struct {
	// contains filtered or unexported fields
}

func (*FSMHandlers) Del

func (h *FSMHandlers) Del(payload *serverpb.RaftLogPayload) error

func (*FSMHandlers) Set

func (h *FSMHandlers) Set(payload *serverpb.RaftLogPayload) error

func (*FSMHandlers) SetWithTTL

func (h *FSMHandlers) SetWithTTL(payload *serverpb.RaftLogPayload) error

func (*FSMHandlers) TrySet

func (h *FSMHandlers) TrySet(payload *serverpb.RaftLogPayload) error

func (*FSMHandlers) TrySetWithTTL

func (h *FSMHandlers) TrySetWithTTL(payload *serverpb.RaftLogPayload) error

type Raft

type Raft struct {
	*raft.Raft
	// contains filtered or unexported fields
}

func NewRaftWithOptions

func NewRaftWithOptions(opts ...RaftServerOption) (*Raft, error)

func (*Raft) AddCluster

func (r *Raft) AddCluster(id raft.ServerID, addr raft.ServerAddress) error

func (*Raft) Term

func (r *Raft) Term() uint64

type RaftApply

type RaftApply interface {
	Apply(ctx *context.Context, m *serverpb.RaftLogPayload, timeout time.Duration) error
}

func NewRaftMultipleLogApply

func NewRaftMultipleLogApply(
	ctx context.Context,
	maxLimit int32,
	deadline,
	applyTimeout time.Duration,
	af ApplyFunc,
) RaftApply

func NewRaftSingeLogApply

func NewRaftSingeLogApply(fc ApplyFunc) RaftApply

type RaftServerOption

type RaftServerOption func(*Raft) error

func RaftWithBoltLogStore

func RaftWithBoltLogStore(path string) RaftServerOption

func RaftWithBootstrap

func RaftWithBootstrap() RaftServerOption

func RaftWithClusters

func RaftWithClusters(clusters []raft.Server) RaftServerOption

func RaftWithConfig

func RaftWithConfig(cfg *raft.Config) RaftServerOption

func RaftWithContext

func RaftWithContext(ctx context.Context) RaftServerOption

func RaftWithEmpty

func RaftWithEmpty() RaftServerOption

func RaftWithFileSnapshotStore

func RaftWithFileSnapshotStore(path string, retain int, logOut io.Writer) RaftServerOption

func RaftWithStdFSM

func RaftWithStdFSM(store store.Store) RaftServerOption

func RaftWithStdStableStore

func RaftWithStdStableStore(store store.Store) RaftServerOption

func RaftWithTCPTransport

func RaftWithTCPTransport(addr string, maxPool int, timeout time.Duration, logOut io.Writer) RaftServerOption

type RaftTLStreamLayer

type RaftTLStreamLayer struct {
	// contains filtered or unexported fields
}

func (RaftTLStreamLayer) Accept

func (r RaftTLStreamLayer) Accept() (net.Conn, error)

func (RaftTLStreamLayer) Addr

func (r RaftTLStreamLayer) Addr() net.Addr

func (RaftTLStreamLayer) Close

func (r RaftTLStreamLayer) Close() error

func (RaftTLStreamLayer) Dial

func (r RaftTLStreamLayer) Dial(addr raft.ServerAddress, timeout time.Duration) (net.Conn, error)

type Server

type Server struct {
	// contains filtered or unexported fields
}

func NewServer

func NewServer(cfg *config.Config) (*Server, error)

func (*Server) ListenAndServe

func (s *Server) ListenAndServe() error

func (*Server) Shutdown

func (s *Server) Shutdown()

type Snapshot

type Snapshot struct {
	io.Reader
}

func (*Snapshot) Persist

func (s *Snapshot) Persist(sink raft.SnapshotSink) error

func (*Snapshot) Release

func (s *Snapshot) Release()

type StableStore

type StableStore struct {
	// contains filtered or unexported fields
}

func NewStableStore

func NewStableStore(s store.Store) (*StableStore, error)

func (*StableStore) Get

func (s *StableStore) Get(key []byte) ([]byte, error)

Get returns the value for key, or an empty byte slice if key was not found.

func (*StableStore) GetUint64

func (s *StableStore) GetUint64(key []byte) (uint64, error)

GetUint64 returns the uint64 value for key, or 0 if key was not found.

func (*StableStore) Set

func (s *StableStore) Set(key []byte, val []byte) error

func (*StableStore) SetUint64

func (s *StableStore) SetUint64(key []byte, val uint64) error