rafthttp

Documentation

Index

• Constants
• Variables
• func NewHandler(p Processor, cid types.ID) http.Handler
• func NewSender(tr http.RoundTripper, u string, cid types.ID, p Processor, ...) *sender
• func NewStreamHandler(finder SenderFinder, id, cid types.ID) http.Handler
• type Batcher
  • func NewBatcher(n int, d time.Duration) *Batcher
  • func (b *Batcher) Reset(t time.Time)
  • func (b *Batcher) ShouldBatch(now time.Time) bool
• type Processor
• type ProposalBatcher
  • func NewProposalBatcher(n int, d time.Duration) *ProposalBatcher
  • func (b *ProposalBatcher) Batch(m raftpb.Message)
  • func (b *ProposalBatcher) IsEmpty() bool
  • func (b *ProposalBatcher) Reset(t time.Time)
• type Sender
• type SenderFinder
• type WriteFlusher

Constants

const (
	ConnReadTimeout  = 5 * time.Second
	ConnWriteTimeout = 5 * time.Second
)

const (
	ConnReadLimitByte = 64 * 1024
)

Variables

var (
	RaftPrefix       = "/raft"
	RaftStreamPrefix = path.Join(RaftPrefix, "stream")
)

Functions

func NewHandler

func NewHandler(p Processor, cid types.ID) http.Handler

func NewSender

func NewSender(tr http.RoundTripper, u string, cid types.ID, p Processor, fs *stats.FollowerStats, shouldstop chan struct{}) *sender

func NewStreamHandler

func NewStreamHandler(finder SenderFinder, id, cid types.ID) http.Handler

NewStreamHandler returns a handler which initiates streamer when receiving stream request from follower.

Types

type Batcher

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

func NewBatcher

func NewBatcher(n int, d time.Duration) *Batcher

func (*Batcher) Reset

func (b *Batcher) Reset(t time.Time)

func (*Batcher) ShouldBatch

func (b *Batcher) ShouldBatch(now time.Time) bool

type Processor

type Processor interface {
	Process(ctx context.Context, m raftpb.Message) error
}

type ProposalBatcher

type ProposalBatcher struct {
	*Batcher
	raftpb.Message
}

func NewProposalBatcher

func NewProposalBatcher(n int, d time.Duration) *ProposalBatcher

func (*ProposalBatcher) Batch

func (b *ProposalBatcher) Batch(m raftpb.Message)

func (*ProposalBatcher) IsEmpty

func (b *ProposalBatcher) IsEmpty() bool

func (*ProposalBatcher) Reset

func (b *ProposalBatcher) Reset(t time.Time)

type Sender

type Sender interface {
	// StartStreaming enables streaming in the sender using the given writer,
	// which provides a fast and efficient way to send appendEntry messages.
	StartStreaming(w WriteFlusher, to types.ID, term uint64) (done <-chan struct{}, err error)
	Update(u string)
	// Send sends the data to the remote node. It is always non-blocking.
	// It may be fail to send data if it returns nil error.
	Send(m raftpb.Message) error
	// Stop performs any necessary finalization and terminates the Sender
	// elegantly.
	Stop()

	// Pause pauses the sender. The sender will simply drops all incoming
	// messages without retruning an error.
	Pause()

	// Resume resumes a paused sender.
	Resume()
}

type SenderFinder

type SenderFinder interface {
	// Sender returns the sender of the given id.
	Sender(id types.ID) Sender
}

type WriteFlusher

type WriteFlusher interface {
	io.Writer
	http.Flusher
}