store

Documentation

Index

• Constants
• type Snapshot
  • func (s *Snapshot[T]) Persist(sink raft.SnapshotSink) error
  • func (s *Snapshot[T]) Release()
• type Store
  • func NewStore[T any](logger *slog.Logger) *Store[T]
  • func (s *Store[T]) Apply(log *raft.Log) interface{}
  • func (s *Store[T]) Initialize(ctx context.Context, conf *consensus.Config) (chan struct{}, error)
  • func (s *Store[T]) Join(nodeID, address string) error
  • func (s *Store[T]) Recieve() (*ds.Message[T], error)
  • func (s *Store[T]) Restore(rc io.ReadCloser) error
  • func (s *Store[T]) Send(data T) error
  • func (s *Store[T]) Snapshot() (raft.FSMSnapshot, error)
  • func (s *Store[T]) Stats() map[string]string
  • func (s *Store[T]) WaitForNodeToBeLeader(duration time.Duration) error

Constants

const (
	Send = iota
	Recieve
)

Specific operations that can be applied to the store

Types

type Snapshot

type Snapshot[T any] struct {
	// contains filtered or unexported fields
}

Snapshot is used to create a snapshot of the queue

func (*Snapshot[T]) Persist

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

Persist is used to persist the snapshot to the sink

func (*Snapshot[T]) Release

func (s *Snapshot[T]) Release()

Release is used to release any resources acquired during the snapshot In this case, we don't have any resources to clean up (noop)

type Store

type Store[T any] struct {
	// contains filtered or unexported fields
}

func NewStore

func NewStore[T any](logger *slog.Logger) *Store[T]

NewStore creates a new store instance with the given logger

func (*Store[T]) Apply

func (s *Store[T]) Apply(log *raft.Log) interface{}

Apply is used to apply a log entry to the store

func (*Store[T]) Initialize

func (s *Store[T]) Initialize(ctx context.Context, conf *consensus.Config) (chan struct{}, error)

Initialize is used to initialize the store with the given config

func (*Store[T]) Join

func (s *Store[T]) Join(nodeID, address string) error

Join is used to join a remote node to the raft cluster

func (*Store[T]) Recieve

func (s *Store[T]) Recieve() (*ds.Message[T], error)

Recieve is used to dequeue a message from the queue

func (*Store[T]) Restore

func (s *Store[T]) Restore(rc io.ReadCloser) error

Restore is used to restore the store from a snapshot

func (*Store[T]) Send

func (s *Store[T]) Send(data T) error

Send is used to enqueue a message into the queue

func (*Store[T]) Snapshot

func (s *Store[T]) Snapshot() (raft.FSMSnapshot, error)

Snapshot is used to create a snapshot of the store

func (*Store[T]) Stats

func (s *Store[T]) Stats() map[string]string

Stats is used to return the stats of the raft instance

func (*Store[T]) WaitForNodeToBeLeader

func (s *Store[T]) WaitForNodeToBeLeader(duration time.Duration) error

WaitForNodeToBeLeader is used to wait for the node to become the leader