lib

Sep 25, 2022

Changes in this version

+ const DefaultUrl

+ const NodeNamePrefix

+ var EntryReplicas = int(0)

+ var StreamNamePrefix = "marmot-changes"

+ var SubjectPrefix = "marmot-change-log"

type ReplicationEvent

+ ChangeRowId int64

+ type Replicator struct

+ func NewReplicator(nodeID uint64, natsServer string, shards uint64) (*Replicator, error)

+ func (r *Replicator) Listen(shardID uint64, callback func(payload []byte) error) error

+ func (r *Replicator) Publish(hash uint64, payload []byte) error

Sep 18, 2022

Changes in this version

+ var MaxLogEntries = uint64(10_000)

type RaftServer

+ func (r *RaftServer) GetSnapshotStateMachine() *SQLiteStateMachine

type SQLiteStateMachine

+ ClusterID uint64

+ func (ssm *SQLiteStateMachine) IsSnapshotEnabled() bool

Sep 11, 2022

Changes in this version

+ const Bootstrap

+ const Entry

+ const Snapshot

+ const State

+ type ControlPane struct

+ func NewControlPane(raft *RaftServer) *ControlPane

+ func (c *ControlPane) Run(addr string) error

+ type RaftServer struct

+ func NewRaftServer(bindAddress string, nodeID uint64, metaPath string, ...) *RaftServer

+ func (r *RaftServer) AddNode(peerID uint64, address string, clusterIDs ...uint64) error

+ func (r *RaftServer) BindCluster(initMembers string, join bool, clusterIDs ...uint64) error

+ func (r *RaftServer) GetActiveClusters() []uint64

+ func (r *RaftServer) GetClusterMap() map[uint64]uint64

+ func (r *RaftServer) GetNodeMap() map[uint64][]uint64

+ func (r *RaftServer) Init() error

+ func (r *RaftServer) LeaderUpdated(info raftio.LeaderInfo)

+ func (r *RaftServer) Propose(key uint64, data []byte, dur time.Duration) (*dragonboat.RequestResult, error)

+ func (r *RaftServer) RequestSnapshot(timeout time.Duration) (uint64, uint64, error)

+ func (r *RaftServer) ShuffleCluster(nodes ...uint64) error

+ func (r *RaftServer) SnapshotCompleted() bool

+ func (r *RaftServer) TransferClusters(toPeerID uint64, clusterIDs ...uint64) error

+ type ReplicationEvent struct

+ FromNodeId uint64

+ Payload *T

+ func (e *ReplicationEvent[T]) Marshal() ([]byte, error)

+ func (e *ReplicationEvent[T]) Unmarshal(data []byte) error

+ type SQLiteLogDB struct

+ func (s *SQLiteLogDB) BinaryFormat() uint32

+ func (s *SQLiteLogDB) Close()

+ func (s *SQLiteLogDB) CompactEntriesTo(clusterID uint64, nodeID uint64, index uint64) (<-chan struct{}, error)

+ func (s *SQLiteLogDB) DeleteSnapshot(clusterID uint64, nodeID uint64, index uint64) error

+ func (s *SQLiteLogDB) GetBootstrapInfo(clusterID uint64, nodeID uint64) (raftpb.Bootstrap, error)

+ func (s *SQLiteLogDB) ImportSnapshot(snp raftpb.Snapshot, nodeID uint64) error

+ func (s *SQLiteLogDB) IterateEntries(entries []raftpb.Entry, size uint64, clusterID uint64, nodeID uint64, ...) ([]raftpb.Entry, uint64, error)

+ func (s *SQLiteLogDB) ListNodeInfo() ([]raftio.NodeInfo, error)

+ func (s *SQLiteLogDB) ListSnapshots(clusterID uint64, nodeID uint64, index uint64) ([]raftpb.Snapshot, error)

+ func (s *SQLiteLogDB) Name() string

+ func (s *SQLiteLogDB) ReadRaftState(clusterID uint64, nodeID uint64, snapshotIndex uint64) (raftio.RaftState, error)

+ func (s *SQLiteLogDB) RemoveEntriesTo(clusterID uint64, nodeID uint64, index uint64) error

+ func (s *SQLiteLogDB) RemoveNodeData(clusterID uint64, nodeID uint64) error

+ func (s *SQLiteLogDB) SaveBootstrapInfo(clusterID uint64, nodeID uint64, bootstrap raftpb.Bootstrap) error

+ func (s *SQLiteLogDB) SaveRaftState(updates []raftpb.Update, _ uint64) error

+ func (s *SQLiteLogDB) SaveSnapshots(updates []raftpb.Update) error

+ type SQLiteLogDBFactory struct

+ func NewSQLiteLogDBFactory(metaPath string, nodeID uint64) *SQLiteLogDBFactory

+ func (f *SQLiteLogDBFactory) Create(_ config.NodeHostConfig, _ config.LogDBCallback, _ []string, _ []string) (raftio.ILogDB, error)

+ func (f *SQLiteLogDBFactory) Name() string

+ type SQLiteStateMachine struct

+ DB *db.SqliteStreamDB

+ NodeID uint64

+ RaftPath string

+ func NewDBStateMachine(nodeID uint64, db *db.SqliteStreamDB, path string) *SQLiteStateMachine

+ func (ssm *SQLiteStateMachine) Close() error

+ func (ssm *SQLiteStateMachine) GetSnapshotDir() (string, error)

+ func (ssm *SQLiteStateMachine) HasRestoredSnapshot() bool

+ func (ssm *SQLiteStateMachine) HasSavedSnapshot() bool

+ func (ssm *SQLiteStateMachine) Lookup(_ interface{}) (interface{}, error)

+ func (ssm *SQLiteStateMachine) Open(_ <-chan struct{}) (uint64, error)

+ func (ssm *SQLiteStateMachine) PrepareSnapshot() (interface{}, error)

+ func (ssm *SQLiteStateMachine) RecoverFromSnapshot(reader io.Reader, _ <-chan struct{}) error

+ func (ssm *SQLiteStateMachine) SaveSnapshot(path interface{}, writer io.Writer, _ <-chan struct{}) error

+ func (ssm *SQLiteStateMachine) Sync() error

+ func (ssm *SQLiteStateMachine) Update(entries []sm.Entry) ([]sm.Entry, error)