kubernetes/vendor/go.etcd.io/etcd/server/v3/etcdserver/api/rafthttp/stream.go

const streamTypeMessage …

const streamTypeMsgAppV2 …

const streamBufSize …

var errUnsupportedStreamType …

var supportedStream …

type streamType …

func (t streamType) endpoint(lg *zap.Logger) string { … }

func (t streamType) String() string { … }

var linkHeartbeatMessage …

func isLinkHeartbeatMessage(m *raftpb.Message) bool { … }

type outgoingConn …

type streamWriter …

// startStreamWriter creates a streamWrite and starts a long running go-routine that accepts
// messages and writes to the attached outgoing connection.
func startStreamWriter(lg *zap.Logger, local, id types.ID, status *peerStatus, fs *stats.FollowerStats, r Raft) *streamWriter { … }

func (cw *streamWriter) run() { … }

func (cw *streamWriter) writec() (chan<- raftpb.Message, bool) { … }

func (cw *streamWriter) close() bool { … }

func (cw *streamWriter) closeUnlocked() bool { … }

func (cw *streamWriter) attach(conn *outgoingConn) bool { … }

func (cw *streamWriter) stop() { … }

type streamReader …

func (cr *streamReader) start() { … }

func (cr *streamReader) run() { … }

func (cr *streamReader) decodeLoop(rc io.ReadCloser, t streamType) error { … }

func (cr *streamReader) stop() { … }

func (cr *streamReader) dial(t streamType) (io.ReadCloser, error) { … }

func (cr *streamReader) close() { … }

func (cr *streamReader) pause() { … }

func (cr *streamReader) resume() { … }

// checkStreamSupport checks whether the stream type is supported in the
// given version.
func checkStreamSupport(v *semver.Version, t streamType) bool { … }