raft

Documentation

Index

• type Command
• type FSM
  • func (f *FSM) Apply(raftLog *raft.Log) interface{}
  • func (f *FSM) Restore(rc io.ReadCloser) error
  • func (f *FSM) Snapshot() (raft.FSMSnapshot, error)
• type FSMResponse
• type FSMSnapshot
  • func (f *FSMSnapshot) Persist(sink raft.SnapshotSink) error
  • func (f *FSMSnapshot) Release()
• type Node
  • func NewNode(db *badger.DB, raftDir string, id, address, raftPort string, peers []string) *Node
  • func (n *Node) Ack(QueueName string, id uint64) error
  • func (n *Node) CreateQueue(queueType, queueName string) error
  • func (n *Node) DeleteQueue(queueName string) error
  • func (n *Node) Dequeue(QueueName string, ack bool) (*queue.Message, error)
  • func (n *Node) Enqueue(queueName string, group string, priority int64, content string) (*queue.Message, error)
  • func (n *Node) GetByID(id uint64) (*queue.Message, error)
  • func (node *Node) Initialize()
  • func (n *Node) IsLeader() bool
  • func (n *Node) Join(nodeID, addr string) error
  • func (n *Node) Leader() string
  • func (node *Node) ListenToLeaderChanges()
  • func (node *Node) SetLeaderChangeFunc(leaderChangeFn func(bool))
  • func (n *Node) UpdatePriority(queueName string, id uint64, priority int64) error

Types

type Command

type Command struct {
	ID        uint64 `json:"id,omitempty"`
	Op        string `json:"op"`
	QueueType string `json:"queue_type"`
	QueueName string `json:"queue_name"`
	Group     string `json:"group"`
	Priority  int64  `json:"priority"`
	Content   string `json:"content"`
	Ack       bool   `json:"ack,omitempty"`
}

type FSM

type FSM struct {
	NodeID string
	// contains filtered or unexported fields
}

func (*FSM) Apply

func (f *FSM) Apply(raftLog *raft.Log) interface{}

func (*FSM) Restore

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

func (*FSM) Snapshot

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

type FSMResponse

type FSMResponse struct {
	QueueName string
	ID        uint64
	Priority  int64
	Content   string
	// contains filtered or unexported fields
}

type FSMSnapshot

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

func (*FSMSnapshot) Persist

func (f *FSMSnapshot) Persist(sink raft.SnapshotSink) error

func (*FSMSnapshot) Release

func (f *FSMSnapshot) Release()

type Node

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

func NewNode

func NewNode(db *badger.DB, raftDir string, id, address, raftPort string, peers []string) *Node

func (*Node) Ack

func (n *Node) Ack(QueueName string, id uint64) error

func (*Node) CreateQueue

func (n *Node) CreateQueue(queueType, queueName string) error

func (*Node) DeleteQueue

func (n *Node) DeleteQueue(queueName string) error

func (*Node) Dequeue

func (n *Node) Dequeue(QueueName string, ack bool) (*queue.Message, error)

func (*Node) Enqueue

func (n *Node) Enqueue(
	queueName string, group string, priority int64, content string,
) (*queue.Message, error)

func (*Node) GetByID

func (n *Node) GetByID(id uint64) (*queue.Message, error)

func (*Node) Initialize

func (node *Node) Initialize()

func (*Node) IsLeader

func (n *Node) IsLeader() bool

func (*Node) Join

func (n *Node) Join(nodeID, addr string) error

func (*Node) Leader

func (n *Node) Leader() string

func (*Node) ListenToLeaderChanges

func (node *Node) ListenToLeaderChanges()

func (*Node) SetLeaderChangeFunc

func (node *Node) SetLeaderChangeFunc(leaderChangeFn func(bool))

func (*Node) UpdatePriority

func (n *Node) UpdatePriority(queueName string, id uint64, priority int64) error