mux

README

Mux is a multiplexing library for golang. It used to combine multi transfer connection into a single one to reduce system load and make connection handshake faster.

Documentation

Index

• Constants
• Variables
• type BufferAllocFunc
• type BufferRecycleFunc
• type Frame
  • func NewFrameContext(ctx context.Context, cmd cmdType, streamId uint32, data []byte) *Frame
  • func (frame *Frame) Close()
  • func (frame *Frame) Cmd() byte
  • func (frame *Frame) MarshalHeader(buffer []byte) (int, error)
  • func (frame *Frame) StreamId() uint32
  • func (frame *Frame) UnMarshalHeader(buffer []byte) (int, error)
• type Logger
• type Option
  • func WithBufferManager(allocFunc BufferAllocFunc, recycleFunc BufferRecycleFunc) Option
  • func WithBufferSize(sizeLimit int) Option
  • func WithFinalizer(finalizer func()) Option
  • func WithHeartBeatInterval(interval time.Duration) Option
  • func WithHeartBeatSwitch(choose bool) Option
  • func WithHeartBeatTTL(ttl time.Duration) Option
  • func WithLogger(logger Logger) Option
  • func WithMetrics(...) Option
  • func WithRole(role roleType) Option
• type Session
  • func NewSession(conn io.ReadWriteCloser, options ...Option) *Session
  • func NewSessionContext(ctx context.Context, conn io.ReadWriteCloser, options ...Option) *Session
  • func (session *Session) AcceptStream(ctx context.Context) (*Stream, error)
  • func (session *Session) Close() error
  • func (session *Session) CloseWithErr(err error)
  • func (session *Session) Ctx() context.Context
  • func (session *Session) Err() error
  • func (session *Session) IsClose() bool
  • func (session *Session) Lifetime() time.Duration
  • func (session *Session) OpenStream(ctx context.Context) (*Stream, error)
  • func (session *Session) StreamCount() int
• type Stream
  • func (stream *Stream) Close() error
  • func (stream *Stream) Done() <-chan struct{}
  • func (stream *Stream) IsClose() bool
  • func (stream *Stream) Lifetime() time.Duration
  • func (stream *Stream) LocalAddr() net.Addr
  • func (stream *Stream) Read(buffer []byte) (int, error)
  • func (stream *Stream) ReadContext(ctx context.Context, buffer []byte) (int, error)
  • func (stream *Stream) RemoteAddr() net.Addr
  • func (stream *Stream) SetDeadline(t time.Time) error
  • func (stream *Stream) SetReadDeadline(t time.Time) error
  • func (stream *Stream) SetWriteDeadline(t time.Time) error
  • func (stream *Stream) Write(buffer []byte) (int, error)
  • func (stream *Stream) WriteContext(ctx context.Context, buffer []byte) (int, error)

Constants

const (
	RoleClient roleType = 1 + iota // client start with odd number, server start with even number
	RoleServer
)

Variables

var (
	BufferSizeLimitErr = errors.New("buffer size limit err")
	UnknownCmdErr      = errors.New("cmd unknown err")
)

var (
	NoopLogger     = &noopLoggerImpl{}
	StandardLogger = &standardLoggerImpl{}
)

var (
	ErrSessionClosed    = errors.New("session closed")
	ErrSessionTTLExceed = errors.New("session ttl exceed")
	ErrStreamIdDup      = errors.New("stream id duplicated err")
	ErrStreamIdNotFound = errors.New("stream id not found")
	ErrBufferLimited    = errors.New("buffer length limited")
)

var (
	ErrStreamClosed = errors.New("stream has been closed")
)

Types

type BufferAllocFunc

type BufferAllocFunc func(size int) []byte

type BufferRecycleFunc

type BufferRecycleFunc func([]byte)

type Frame

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

func NewFrameContext

func NewFrameContext(ctx context.Context, cmd cmdType, streamId uint32, data []byte) *Frame

func (*Frame) Close

func (frame *Frame) Close()

func (*Frame) Cmd

func (frame *Frame) Cmd() byte

func (*Frame) MarshalHeader

func (frame *Frame) MarshalHeader(buffer []byte) (int, error)

func (*Frame) StreamId

func (frame *Frame) StreamId() uint32

func (*Frame) UnMarshalHeader

func (frame *Frame) UnMarshalHeader(buffer []byte) (int, error)

type Logger

type Logger interface {
	Debugf(_ string, _ ...interface{})
	Infof(_ string, _ ...interface{})
	Warnf(_ string, _ ...interface{})
	Errorf(_ string, _ ...interface{})
}

type Option

type Option func(*Session)

func WithBufferManager

func WithBufferManager(allocFunc BufferAllocFunc, recycleFunc BufferRecycleFunc) Option

func WithBufferSize

func WithBufferSize(sizeLimit int) Option

func WithFinalizer

func WithFinalizer(finalizer func()) Option

func WithHeartBeatInterval

func WithHeartBeatInterval(interval time.Duration) Option

func WithHeartBeatSwitch

func WithHeartBeatSwitch(choose bool) Option

func WithHeartBeatTTL

func WithHeartBeatTTL(ttl time.Duration) Option

func WithLogger

func WithLogger(logger Logger) Option

func WithMetrics

func WithMetrics(writeFrameDurations, recvFrameDurations, acceptStreamDurations prometheus.Histogram) Option

func WithRole

func WithRole(role roleType) Option

type Session

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

func NewSession

func NewSession(conn io.ReadWriteCloser, options ...Option) *Session

func NewSessionContext

func NewSessionContext(ctx context.Context, conn io.ReadWriteCloser, options ...Option) *Session

func (*Session) AcceptStream

func (session *Session) AcceptStream(ctx context.Context) (*Stream, error)

func (*Session) Close

func (session *Session) Close() error

func (*Session) CloseWithErr

func (session *Session) CloseWithErr(err error)

func (*Session) Ctx

func (session *Session) Ctx() context.Context

func (*Session) Err

func (session *Session) Err() error

func (*Session) IsClose

func (session *Session) IsClose() bool

func (*Session) Lifetime

func (session *Session) Lifetime() time.Duration

func (*Session) OpenStream

func (session *Session) OpenStream(ctx context.Context) (*Stream, error)

OpenStream create a new established stream connection

func (*Session) StreamCount

func (session *Session) StreamCount() int

type Stream

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

func (*Stream) Close

func (stream *Stream) Close() error

Close positive close, need to notify remote

func (*Stream) Done

func (stream *Stream) Done() <-chan struct{}

func (*Stream) IsClose

func (stream *Stream) IsClose() bool

func (*Stream) Lifetime

func (stream *Stream) Lifetime() time.Duration

func (*Stream) LocalAddr

func (stream *Stream) LocalAddr() net.Addr

func (*Stream) Read

func (stream *Stream) Read(buffer []byte) (int, error)

func (*Stream) ReadContext

func (stream *Stream) ReadContext(ctx context.Context, buffer []byte) (int, error)

func (*Stream) RemoteAddr

func (stream *Stream) RemoteAddr() net.Addr

func (*Stream) SetDeadline

func (stream *Stream) SetDeadline(t time.Time) error

func (*Stream) SetReadDeadline

func (stream *Stream) SetReadDeadline(t time.Time) error

func (*Stream) SetWriteDeadline

func (stream *Stream) SetWriteDeadline(t time.Time) error

func (*Stream) Write

func (stream *Stream) Write(buffer []byte) (int, error)

func (*Stream) WriteContext

func (stream *Stream) WriteContext(ctx context.Context, buffer []byte) (int, error)