raft

Documentation

Index

• type AckPayload
• type Command
  • func (c Command) MarshalJSON() ([]byte, error)
  • func (c *Command) UnmarshalJSON(data []byte) error
• type CreateQueuePayload
• type DeleteQueuePayload
• type DequeuePayload
• type EnqueuePayload
• 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 NackPayload
• type Node
  • func NewNode(db *badger.DB, raftDir string, cfg *config.Config, 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 (node *Node) GenerateID() uint64
  • func (n *Node) GetByID(id uint64) (*queue.Message, error)
  • func (n *Node) GetQueueInfo(queueName string) (*queue.QueueInfo, error)
  • func (n *Node) GetQueues() []*queue.QueueInfo
  • func (node *Node) InitIDGenerator() 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 (n *Node) Nack(QueueName string, id uint64) error
  • func (node *Node) PrometheusRegistry() prometheus.Registerer
  • func (node *Node) SetLeaderChangeFunc(leaderChangeFn func(bool))
  • func (n *Node) UpdatePriority(queueName string, id uint64, priority int64) error
• type UpdatePriorityPayload

Types

type AckPayload

type AckPayload struct {
	QueueName string `json:"queue_name"`
	ID        uint64 `json:"id"`
}

type Command

type Command struct {
	Op      string      `json:"op"`
	Payload interface{} `json:"payload"`
}

func (Command) MarshalJSON

func (c Command) MarshalJSON() ([]byte, error)

Custom marshal logic to encode the dynamic payload

func (*Command) UnmarshalJSON

func (c *Command) UnmarshalJSON(data []byte) error

Custom unmarshal logic to decode the dynamic payload based on the "op" field

type CreateQueuePayload

type CreateQueuePayload struct {
	QueueType string `json:"queue_type"`
	QueueName string `json:"queue_name"`
}

type DeleteQueuePayload

type DeleteQueuePayload struct {
	QueueName string `json:"queue_name"`
}

type DequeuePayload

type DequeuePayload struct {
	QueueName string `json:"queue_name"`
	Ack       bool   `json:"ack"`
}

type EnqueuePayload

type EnqueuePayload struct {
	ID        uint64 `json:"id"`
	QueueName string `json:"queue_name"`
	Group     string `json:"group"`
	Priority  int64  `json:"priority"`
	Content   string `json:"content"`
}

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
	Group     string
	Priority  int64
	Content   string
	// contains filtered or unexported fields
}

type FSMSnapshot

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

func (*FSMSnapshot) Persist

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

func (*FSMSnapshot) Release

func (f *FSMSnapshot) Release()

type NackPayload

type NackPayload struct {
	QueueName string `json:"queue_name"`
	ID        uint64 `json:"id"`
}

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, cfg *config.Config, 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) GenerateID

func (node *Node) GenerateID() uint64

func (*Node) GetByID

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

func (*Node) GetQueueInfo

func (n *Node) GetQueueInfo(queueName string) (*queue.QueueInfo, error)

func (*Node) GetQueues

func (n *Node) GetQueues() []*queue.QueueInfo

func (*Node) InitIDGenerator

func (node *Node) InitIDGenerator() 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) Nack

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

func (*Node) PrometheusRegistry

func (node *Node) PrometheusRegistry() prometheus.Registerer

func (*Node) SetLeaderChangeFunc

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

func (*Node) UpdatePriority

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

type UpdatePriorityPayload

type UpdatePriorityPayload struct {
	QueueName string `json:"queue_name"`
	ID        uint64 `json:"id"`
	Priority  int64  `json:"priority"`
}