...
ActiveGo 1.8Package rafthttp
Overview ▹
Index ▹
Constants
const (
// ConnReadTimeout and ConnWriteTimeout are the i/o timeout set on each connection rafthttp pkg creates.
// A 5 seconds timeout is good enough for recycling bad connections. Or we have to wait for
// tcp keepalive failing to detect a bad connection, which is at minutes level.
// For long term streaming connections, rafthttp pkg sends application level linkHeartbeatMessage
// to keep the connection alive.
// For short term pipeline connections, the connection MUST be killed to avoid it being
// put back to http pkg connection pool.
ConnReadTimeout = 5 * time.Second
ConnWriteTimeout = 5 * time.Second
)
Variables
var (
RaftPrefix = "/raft"
ProbingPrefix = path.Join(RaftPrefix, "probing")
RaftStreamPrefix = path.Join(RaftPrefix, "stream")
RaftSnapshotPrefix = path.Join(RaftPrefix, "snapshot")
)
var (
ErrExceedSizeLimit = errors.New("rafthttp: error limit exceeded")
)
func NewListener ¶
func NewListener(u url.URL, tlsinfo *transport.TLSInfo) (net.Listener, error)
NewListener returns a listener for raft message transfer between peers. It uses timeout listener to identify broken streams promptly.
func NewRoundTripper ¶
func NewRoundTripper(tlsInfo transport.TLSInfo, dialTimeout time.Duration) (http.RoundTripper, error)
NewRoundTripper returns a roundTripper used to send requests to rafthttp listener of remote peers.
type Pausable ¶
Pausable is a testing interface for pausing transport traffic.
type Pausable interface {
Pause()
Resume()
}
type Peer ¶
type Peer interface {
// contains filtered or unexported methods
}
type Raft ¶
type Raft interface {
Process(ctx context.Context, m raftpb.Message) error
IsIDRemoved(id uint64) bool
ReportUnreachable(id uint64)
ReportSnapshot(id uint64, status raft.SnapshotStatus)
}
type Transport ¶
Transport implements Transporter interface. It provides the functionality to send raft messages to peers, and receive raft messages from peers. User should call Handler method to get a handler to serve requests received from peerURLs. User needs to call Start before calling other functions, and call Stop when the Transport is no longer used.
type Transport struct {
DialTimeout time.Duration // maximum duration before timing out dial of the request
TLSInfo transport.TLSInfo // TLS information used when creating connection
ID types.ID // local member ID
URLs types.URLs // local peer URLs
ClusterID types.ID // raft cluster ID for request validation
Raft Raft // raft state machine, to which the Transport forwards received messages and reports status
Snapshotter *snap.Snapshotter
ServerStats *stats.ServerStats // used to record general transportation statistics
// used to record transportation statistics with followers when
// performing as leader in raft protocol
LeaderStats *stats.LeaderStats
// ErrorC is used to report detected critical errors, e.g.,
// the member has been permanently removed from the cluster
// When an error is received from ErrorC, user should stop raft state
// machine and thus stop the Transport.
ErrorC chan error
// contains filtered or unexported fields
}
func (*Transport) ActiveSince ¶
func (t *Transport) ActiveSince(id types.ID) time.Time
func (*Transport) AddPeer ¶
func (t *Transport) AddPeer(id types.ID, us []string)
func (*Transport) AddRemote ¶
func (t *Transport) AddRemote(id types.ID, us []string)
func (*Transport) CutPeer ¶
func (t *Transport) CutPeer(id types.ID)
CutPeer drops messages to the specified peer.
func (*Transport) Get ¶
func (t *Transport) Get(id types.ID) Peer
func (*Transport) Handler ¶
func (t *Transport) Handler() http.Handler
func (*Transport) MendPeer ¶
func (t *Transport) MendPeer(id types.ID)
MendPeer recovers the message dropping behavior of the given peer.
func (*Transport) Pause ¶
func (t *Transport) Pause()
func (*Transport) RemoveAllPeers ¶
func (t *Transport) RemoveAllPeers()
func (*Transport) RemovePeer ¶
func (t *Transport) RemovePeer(id types.ID)
func (*Transport) Resume ¶
func (t *Transport) Resume()
func (*Transport) Send ¶
func (t *Transport) Send(msgs []raftpb.Message)
func (*Transport) SendSnapshot ¶
func (t *Transport) SendSnapshot(m snap.Message)
func (*Transport) Start ¶
func (t *Transport) Start() error
func (*Transport) Stop ¶
func (t *Transport) Stop()
func (*Transport) UpdatePeer ¶
func (t *Transport) UpdatePeer(id types.ID, us []string)
type Transporter ¶
type Transporter interface {
// Start starts the given Transporter.
// Start MUST be called before calling other functions in the interface.
Start() error
// Handler returns the HTTP handler of the transporter.
// A transporter HTTP handler handles the HTTP requests
// from remote peers.
// The handler MUST be used to handle RaftPrefix(/raft)
// endpoint.
Handler() http.Handler
// Send sends out the given messages to the remote peers.
// Each message has a To field, which is an id that maps
// to an existing peer in the transport.
// If the id cannot be found in the transport, the message
// will be ignored.
Send(m []raftpb.Message)
// SendSnapshot sends out the given snapshot message to a remote peer.
// The behavior of SendSnapshot is similar to Send.
SendSnapshot(m snap.Message)
// AddRemote adds a remote with given peer urls into the transport.
// A remote helps newly joined member to catch up the progress of cluster,
// and will not be used after that.
// It is the caller's responsibility to ensure the urls are all valid,
// or it panics.
AddRemote(id types.ID, urls []string)
// AddPeer adds a peer with given peer urls into the transport.
// It is the caller's responsibility to ensure the urls are all valid,
// or it panics.
// Peer urls are used to connect to the remote peer.
AddPeer(id types.ID, urls []string)
// RemovePeer removes the peer with given id.
RemovePeer(id types.ID)
// RemoveAllPeers removes all the existing peers in the transport.
RemoveAllPeers()
// UpdatePeer updates the peer urls of the peer with the given id.
// It is the caller's responsibility to ensure the urls are all valid,
// or it panics.
UpdatePeer(id types.ID, urls []string)
// ActiveSince returns the time that the connection with the peer
// of the given id becomes active.
// If the connection is active since peer was added, it returns the adding time.
// If the connection is currently inactive, it returns zero time.
ActiveSince(id types.ID) time.Time
// Stop closes the connections and stops the transporter.
Stop()
}
func NewNopTransporter ¶
func NewNopTransporter() Transporter
func NewSnapTransporter ¶
func NewSnapTransporter(snapDir string) (Transporter, <-chan snap.Message)