ha

Documentation

Index

• Constants
• func NewFSM(base db.Provider) raft.FSM
• func NewProvider(r *raft.Raft, base db.Provider) db.Provider
• type DB
  • func NewDB(db *badger.DB) *DB
  • func (db *DB) DeleteRange(min, max uint64) error
  • func (db *DB) FirstIndex() (v uint64, err error)
  • func (db *DB) Get(key []byte) (v []byte, err error)
  • func (db *DB) GetLog(index uint64, log *raft.Log) error
  • func (db *DB) GetUint64(key []byte) (uint64, error)
  • func (db *DB) LastIndex() (v uint64, err error)
  • func (db *DB) Set(key, value []byte) error
  • func (db *DB) SetUint64(key []byte, value uint64) error
  • func (db *DB) StoreLog(log *raft.Log) error
  • func (db *DB) StoreLogs(logs []*raft.Log) error
• type Ha
  • func Get(ctx context.Context) *Ha
  • func Open(ctx context.Context) (context.Context, *Ha)
  • func (h *Ha) Close() error
• type ProtoTransit
  • func (t *ProtoTransit) AppendEntries(ctx context.Context, req *raftv1.AppendEntriesRequest) (*raftv1.AppendEntriesResponse, error)
  • func (t *ProtoTransit) AppendEntriesPipeline(s raftv1.Transport_AppendEntriesPipelineServer) error
  • func (t *ProtoTransit) InstallSnapshot(s raftv1.Transport_InstallSnapshotServer) error
  • func (t *ProtoTransit) RequestVote(ctx context.Context, req *raftv1.RequestVoteRequest) (*raftv1.RequestVoteResponse, error)
  • func (t *ProtoTransit) TimeoutNow(ctx context.Context, req *raftv1.TimeoutNowRequest) (*raftv1.TimeoutNowResponse, error)
• type RPCRequest
• type RPCResponse
• type Transit
  • func NewTransport(ctx context.Context, addr raft.ServerAddress, timeout time.Duration) *Transit
  • func (t *Transit) AppendEntries(id raft.ServerID, target raft.ServerAddress, args *raft.AppendEntriesRequest, ...) error
  • func (t *Transit) AppendEntriesPipeline(id raft.ServerID, target raft.ServerAddress) (raft.AppendPipeline, error)
  • func (t *Transit) Close() error
  • func (t *Transit) Consumer() <-chan raft.RPC
  • func (*Transit) DecodePeer(buf []byte) raft.ServerAddress
  • func (*Transit) EncodePeer(id raft.ServerID, p raft.ServerAddress) []byte
  • func (t *Transit) InstallSnapshot(id raft.ServerID, target raft.ServerAddress, args *raft.InstallSnapshotRequest, ...) error
  • func (t *Transit) LocalAddr() raft.ServerAddress
  • func (t *Transit) Register(s *grpc.Server)
  • func (t *Transit) RequestVote(id raft.ServerID, target raft.ServerAddress, args *raft.RequestVoteRequest, ...) error
  • func (t *Transit) SetHeartbeatHandler(cb func(rpc raft.RPC))
  • func (t *Transit) TimeoutNow(id raft.ServerID, target raft.ServerAddress, args *raft.TimeoutNowRequest, ...) error

Constants

const ApplyTTL = time.Second

Functions

func NewFSM

func NewFSM(base db.Provider) raft.FSM

func NewProvider

func NewProvider(r *raft.Raft, base db.Provider) db.Provider

NewProvider returns a new db.Provider that uses raft to distribute writes. Reads are always local to the base provider.

Types

type DB

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

func NewDB

func NewDB(db *badger.DB) *DB

func (*DB) DeleteRange

func (db *DB) DeleteRange(min, max uint64) error

func (*DB) FirstIndex

func (db *DB) FirstIndex() (v uint64, err error)

func (*DB) Get

func (db *DB) Get(key []byte) (v []byte, err error)

func (*DB) GetLog

func (db *DB) GetLog(index uint64, log *raft.Log) error

func (*DB) GetUint64

func (db *DB) GetUint64(key []byte) (uint64, error)

func (*DB) LastIndex

func (db *DB) LastIndex() (v uint64, err error)

func (*DB) Set

func (db *DB) Set(key, value []byte) error

func (*DB) SetUint64

func (db *DB) SetUint64(key []byte, value uint64) error

func (*DB) StoreLog

func (db *DB) StoreLog(log *raft.Log) error

func (*DB) StoreLogs

func (db *DB) StoreLogs(logs []*raft.Log) error

type Ha

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

func Get

func Get(ctx context.Context) *Ha

func Open

func Open(ctx context.Context) (context.Context, *Ha)

func (*Ha) Close

func (h *Ha) Close() error

type ProtoTransit

type ProtoTransit struct {
	raftv1.UnsafeTransportServer
	// contains filtered or unexported fields
}

func (*ProtoTransit) AppendEntries

func (t *ProtoTransit) AppendEntries(ctx context.Context, req *raftv1.AppendEntriesRequest) (*raftv1.AppendEntriesResponse, error)

func (*ProtoTransit) AppendEntriesPipeline

func (t *ProtoTransit) AppendEntriesPipeline(s raftv1.Transport_AppendEntriesPipelineServer) error

func (*ProtoTransit) InstallSnapshot

func (t *ProtoTransit) InstallSnapshot(s raftv1.Transport_InstallSnapshotServer) error

func (*ProtoTransit) RequestVote

func (t *ProtoTransit) RequestVote(ctx context.Context, req *raftv1.RequestVoteRequest) (*raftv1.RequestVoteResponse, error)

func (*ProtoTransit) TimeoutNow

func (t *ProtoTransit) TimeoutNow(ctx context.Context, req *raftv1.TimeoutNowRequest) (*raftv1.TimeoutNowResponse, error)

type RPCRequest

type RPCRequest interface {
	*raft.AppendEntriesRequest |
		*raft.RequestVoteRequest |
		*raft.InstallSnapshotRequest |
		*raft.TimeoutNowRequest
}

type RPCResponse

type RPCResponse interface {
	*raft.AppendEntriesResponse |
		*raft.RequestVoteResponse |
		*raft.InstallSnapshotResponse |
		*raft.TimeoutNowResponse
}

type Transit

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

func NewTransport

func NewTransport(ctx context.Context, addr raft.ServerAddress, timeout time.Duration) *Transit

func (*Transit) AppendEntries

func (t *Transit) AppendEntries(
	id raft.ServerID,
	target raft.ServerAddress,
	args *raft.AppendEntriesRequest, resp *raft.AppendEntriesResponse) error

func (*Transit) AppendEntriesPipeline

func (t *Transit) AppendEntriesPipeline(id raft.ServerID, target raft.ServerAddress) (raft.AppendPipeline, error)

func (*Transit) Close

func (t *Transit) Close() error

func (*Transit) Consumer

func (t *Transit) Consumer() <-chan raft.RPC

func (*Transit) DecodePeer

func (*Transit) DecodePeer(buf []byte) raft.ServerAddress

func (*Transit) EncodePeer

func (*Transit) EncodePeer(id raft.ServerID, p raft.ServerAddress) []byte

func (*Transit) InstallSnapshot

func (t *Transit) InstallSnapshot(id raft.ServerID, target raft.ServerAddress, args *raft.InstallSnapshotRequest, resp *raft.InstallSnapshotResponse, data io.Reader) error

func (*Transit) LocalAddr

func (t *Transit) LocalAddr() raft.ServerAddress

func (*Transit) Register

func (t *Transit) Register(s *grpc.Server)

func (*Transit) RequestVote

func (t *Transit) RequestVote(id raft.ServerID,
	target raft.ServerAddress,
	args *raft.RequestVoteRequest, resp *raft.RequestVoteResponse) error

func (*Transit) SetHeartbeatHandler

func (t *Transit) SetHeartbeatHandler(cb func(rpc raft.RPC))

func (*Transit) TimeoutNow

func (t *Transit) TimeoutNow(id raft.ServerID, target raft.ServerAddress, args *raft.TimeoutNowRequest, resp *raft.TimeoutNowResponse) error