mux

Documentation

Index

• Constants
• Variables
• func DecodeFrame(raw []byte) (StreamID, FrameType, []byte, error)
• func EncodeFrame(streamID StreamID, ftype FrameType, payload []byte) []byte
• type FrameType
• type Metrics
• type Mux
  • func New(conn xfer.Conn, role Role) *Mux
  • func NewWithOpts(conn xfer.Conn, role Role, opts ...Option) *Mux
  • func (m *Mux) Accept(ctx context.Context) (Stream, error)
  • func (m *Mux) Close() error
  • func (m *Mux) Context() context.Context
  • func (m *Mux) Done() <-chan struct{}
  • func (m *Mux) Metrics() *Metrics
  • func (m *Mux) Open(ctx context.Context) (Stream, error)
  • func (m *Mux) Role() Role
• type Option
  • func WithAcceptChSize(n int) Option
  • func WithMaxStreams(n int) Option
• type Role
• type Stream
• type StreamID
• type StreamMetrics

Constants

const DefaultWindowSize = 65536 // 64 KB

DefaultWindowSize 是每条流的初始发送窗口大小。

const FrameHeaderSize = 8

FrameHeaderSize 是帧头部的固定字节数。

Variables

var (
	// ErrFrameTooShort 是帧数据不足 FrameHeaderSize 时的错误。
	ErrFrameTooShort = errors.New("mux: frame too short")
	// ErrFrameTruncated 是帧负载长度少于声明值时的错误。
	ErrFrameTruncated = errors.New("mux: frame payload truncated")
)

var (
	ErrStreamRejected = errors.New("mux: stream rejected")
	ErrMaxStreams     = errors.New("mux: max streams reached")
)

Sentinel errors.

Functions

func DecodeFrame

func DecodeFrame(raw []byte) (StreamID, FrameType, []byte, error)

DecodeFrame 解码一个完整帧。

func EncodeFrame

func EncodeFrame(streamID StreamID, ftype FrameType, payload []byte) []byte

EncodeFrame 编码一个完整帧。

Types

type FrameType

type FrameType byte

FrameType 标识帧的用途。

const (
	FrameData         FrameType = 0x00 // 用户流数据
	FrameOpen         FrameType = 0x01 // 通知远端打开新流
	FrameClose        FrameType = 0x02 // 关闭指定流
	FramePing         FrameType = 0x03 // 心跳探测
	FramePong         FrameType = 0x04 // 心跳回复
	FrameCloseWrite   FrameType = 0x05 // 写半关闭（不再有更多数据发送）
	FrameReject       FrameType = 0x06 // 拒绝流创建（acceptCh 满或达到 maxStreams）
	FrameWindowUpdate FrameType = 0x07 // 窗口更新（流控）
)

type Metrics

type Metrics struct {
	Streams               StreamMetrics
	PingsSent             atomic.Int64
	PongsReceived         atomic.Int64
	FramesReceived        atomic.Int64
	FramesSent            atomic.Int64
	Errors                atomic.Int64
	StreamsRejected       atomic.Int64 // 因 acceptCh 满或 maxStreams 限制被拒绝的流数
	RecvRetries           atomic.Int64 // 读取循环重试次数
	StreamsRejectedAccCh  atomic.Int64 // 因 acceptCh 满被拒绝的流数
	StreamsRejectedMaxStr atomic.Int64 // 因 maxStreams 被拒绝的流数
}

Metrics 收集 mux 级别的统计信息。

type Mux

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

Mux 在一条 xfer.Conn 上多路复用多条虚拟流。

func New

func New(conn xfer.Conn, role Role) *Mux

New 创建 Mux，启动事件循环 goroutine。

func NewWithOpts

func NewWithOpts(conn xfer.Conn, role Role, opts ...Option) *Mux

NewWithOpts 创建 Mux 并应用选项。

func (*Mux) Accept

func (m *Mux) Accept(ctx context.Context) (Stream, error)

Accept 等待并返回一条新流。

func (*Mux) Close

func (m *Mux) Close() error

Close 关闭 mux 和所有流。

func (*Mux) Context

func (m *Mux) Context() context.Context

func (*Mux) Done

func (m *Mux) Done() <-chan struct{}

Done 返回一个 channel，当 mux 关闭时关闭（用于测试）。

func (*Mux) Metrics

func (m *Mux) Metrics() *Metrics

Metrics 返回指向 mux 统计信息的指针。

func (*Mux) Open

func (m *Mux) Open(ctx context.Context) (Stream, error)

Open 创建一条新流。

func (*Mux) Role

func (m *Mux) Role() Role

Role 返回 mux 的角色（RoleDialer 或 RoleListener）。

type Option

type Option func(*Mux)

Option 配置 Mux 的函数选项。

func WithAcceptChSize

func WithAcceptChSize(n int) Option

WithAcceptChSize 设置 acceptCh 缓冲区大小，默认 64。

func WithMaxStreams

func WithMaxStreams(n int) Option

WithMaxStreams 设置最大并发流数。

type Role

type Role int

Role 标识 Mux 的角色。

const (
	RoleDialer Role = iota
	RoleListener
)

type Stream

type Stream interface {
	io.ReadWriteCloser
	ID() StreamID
	CloseWrite() error
}

Stream 是虚拟流接口，实现 io.ReadWriteCloser 并支持半关闭。

type StreamID

type StreamID uint32

StreamID 是虚拟流标识符。 控制流使用 StreamID=0，用户流从 1 开始。

type StreamMetrics

type StreamMetrics struct {
	Opened       atomic.Int64
	Closed       atomic.Int64
	BytesRead    atomic.Int64
	BytesWritten atomic.Int64
	Errors       atomic.Int64
}

StreamMetrics 收集流的统计信息。