README
¶
Yamux
Yamux (Yet another Multiplexer) is a multiplexing library for Golang. It relies on an underlying connection to provide reliability and ordering, such as TCP or Unix domain sockets, and provides stream-oriented multiplexing. It is inspired by SPDY but is not interoperable with it.
Yamux features include:
- Bi-directional streams
- Streams can be opened by either client or server
- Useful for NAT traversal
- Server-side push support
- Flow control
- Avoid starvation
- Back-pressure to prevent overwhelming a receiver
- Keep Alives
- Enables persistent connections over a load balancer
- Efficient
- Enables thousands of logical streams with low overhead
Documentation
For complete documentation, see pkg.go.dev.
Specification
The full specification for Yamux is provided in the spec.md file.
It can be used as a guide to implementors of interoperable libraries.
Usage
Using Yamux is remarkably simple:
func client() {
// Get a TCP connection
conn, err := net.Dial(...)
if err != nil {
panic(err)
}
// Setup client side of yamux
session, err := yamux.Client(conn, nil)
if err != nil {
panic(err)
}
// Open a new stream
stream, err := session.Open()
if err != nil {
panic(err)
}
// Stream implements net.Conn
stream.Write([]byte("ping"))
}
func server() {
// Accept a TCP connection
conn, err := listener.Accept()
if err != nil {
panic(err)
}
// Setup server side of yamux
session, err := yamux.Server(conn, nil)
if err != nil {
panic(err)
}
// Accept a stream
stream, err := session.Accept()
if err != nil {
panic(err)
}
// Listen for a message
buf := make([]byte, 4)
stream.Read(buf)
}
Documentation
¶
Index ¶
- Variables
- func VerifyConfig(config *Config) error
- type Config
- type Logger
- type NetError
- type Session
- func (s *Session) Accept() (net.Conn, error)
- func (s *Session) AcceptStream() (*Stream, error)
- func (s *Session) AcceptStreamWithContext(ctx context.Context) (*Stream, error)
- func (s *Session) Addr() net.Addr
- func (s *Session) Close() error
- func (s *Session) CloseChan() <-chan struct{}
- func (s *Session) GoAway() error
- func (s *Session) IsClosed() bool
- func (s *Session) LocalAddr() net.Addr
- func (s *Session) NumStreams() int
- func (s *Session) Open() (net.Conn, error)
- func (s *Session) OpenStream() (*Stream, error)
- func (s *Session) Ping() (time.Duration, error)
- func (s *Session) RemoteAddr() net.Addr
- type Stream
- func (s *Stream) Close() error
- func (s *Stream) LocalAddr() net.Addr
- func (s *Stream) Read(b []byte) (n int, err error)
- func (s *Stream) RemoteAddr() net.Addr
- func (s *Stream) Session() *Session
- func (s *Stream) SetDeadline(t time.Time) error
- func (s *Stream) SetReadDeadline(t time.Time) error
- func (s *Stream) SetWriteDeadline(t time.Time) error
- func (s *Stream) Shrink()
- func (s *Stream) StreamID() uint32
- func (s *Stream) Write(b []byte) (n int, err error)
Constants ¶
This section is empty.
Variables ¶
var ( // ErrInvalidVersion means we received a frame with an // invalid version ErrInvalidVersion = fmt.Errorf("invalid protocol version") // ErrInvalidMsgType means we received a frame with an // invalid message type ErrInvalidMsgType = fmt.Errorf("invalid msg type") // ErrSessionShutdown is used if there is a shutdown during // an operation ErrSessionShutdown = fmt.Errorf("session shutdown") // ErrStreamsExhausted is returned if we have no more // stream ids to issue ErrStreamsExhausted = fmt.Errorf("streams exhausted") // ErrDuplicateStream is used if a duplicate stream is // opened inbound ErrDuplicateStream = fmt.Errorf("duplicate stream initiated") // ErrReceiveWindowExceeded indicates the window was exceeded ErrRecvWindowExceeded = fmt.Errorf("recv window exceeded") // ErrTimeout is used when we reach an IO deadline ErrTimeout = &NetError{ err: fmt.Errorf("i/o deadline reached"), timeout: true, temporary: true, } // ErrStreamClosed is returned when using a closed stream ErrStreamClosed = fmt.Errorf("stream closed") // ErrUnexpectedFlag is set when we get an unexpected flag ErrUnexpectedFlag = fmt.Errorf("unexpected flag") // ErrRemoteGoAway is used when we get a go away from the other side ErrRemoteGoAway = fmt.Errorf("remote end is not accepting connections") // ErrConnectionReset is sent if a stream is reset. This can happen // if the backlog is exceeded, or if there was a remote GoAway. ErrConnectionReset = fmt.Errorf("connection reset") // ErrConnectionWriteTimeout indicates that we hit the "safety valve" // timeout writing to the underlying stream connection. ErrConnectionWriteTimeout = fmt.Errorf("connection write timeout") // ErrKeepAliveTimeout is sent if a missed keepalive caused the stream close ErrKeepAliveTimeout = fmt.Errorf("keepalive timeout") )
Functions ¶
func VerifyConfig ¶
VerifyConfig is used to verify the sanity of configuration
Types ¶
type Config ¶
type Config struct {
// AcceptBacklog is used to limit how many streams may be
// waiting an accept.
AcceptBacklog int
// EnableKeepalive is used to do a period keep alive
// messages using a ping.
EnableKeepAlive bool
// KeepAliveInterval is how often to perform the keep alive
KeepAliveInterval time.Duration
// ConnectionWriteTimeout is meant to be a "safety valve" timeout after
// we which will suspect a problem with the underlying connection and
// close it. This is only applied to writes, where's there's generally
// an expectation that things will move along quickly.
ConnectionWriteTimeout time.Duration
// MaxStreamWindowSize is used to control the maximum
// window size that we allow for a stream.
MaxStreamWindowSize uint32
// StreamOpenTimeout is the maximum amount of time that a stream will
// be allowed to remain in pending state while waiting for an ack from the peer.
// Once the timeout is reached the session will be gracefully closed.
// A zero value disables the StreamOpenTimeout allowing unbounded
// blocking on OpenStream calls.
StreamOpenTimeout time.Duration
// StreamCloseTimeout is the maximum time that a stream will allowed to
// be in a half-closed state when `Close` is called before forcibly
// closing the connection. Forcibly closed connections will empty the
// receive buffer, drop any future packets received for that stream,
// and send a RST to the remote side.
StreamCloseTimeout time.Duration
// LogOutput is used to control the log destination. Either Logger or
// LogOutput can be set, not both.
LogOutput io.Writer
// Logger is used to pass in the logger to be used. Either Logger or
// LogOutput can be set, not both.
Logger Logger
}
Config is used to tune the Yamux session
func DefaultConfig ¶
func DefaultConfig() *Config
DefaultConfig is used to return a default configuration
type Logger ¶
type Logger interface {
Print(v ...interface{})
Printf(format string, v ...interface{})
Println(v ...interface{})
}
Logger is a abstract of *log.Logger
type NetError ¶
type NetError struct {
// contains filtered or unexported fields
}
NetError implements net.Error
type Session ¶
type Session struct {
// contains filtered or unexported fields
}
Session is used to wrap a reliable ordered connection and to multiplex it into multiple streams.
func Client ¶
func Client(conn io.ReadWriteCloser, config *Config) (*Session, error)
Client is used to initialize a new client-side connection. There must be at most one client-side connection.
func Server ¶
func Server(conn io.ReadWriteCloser, config *Config) (*Session, error)
Server is used to initialize a new server-side connection. There must be at most one server-side connection. If a nil config is provided, the DefaultConfiguration will be used.
func (*Session) Accept ¶
Accept is used to block until the next available stream is ready to be accepted.
func (*Session) AcceptStream ¶
AcceptStream is used to block until the next available stream is ready to be accepted.
func (*Session) AcceptStreamWithContext ¶
AcceptStream is used to block until the next available stream is ready to be accepted.
func (*Session) Close ¶
Close is used to close the session and all streams. Attempts to send a GoAway before closing the connection.
func (*Session) CloseChan ¶
func (s *Session) CloseChan() <-chan struct{}
CloseChan returns a read-only channel which is closed as soon as the session is closed.
func (*Session) GoAway ¶
GoAway can be used to prevent accepting further connections. It does not close the underlying conn.
func (*Session) LocalAddr ¶
LocalAddr is used to get the local address of the underlying connection.
func (*Session) NumStreams ¶
NumStreams returns the number of currently open streams
func (*Session) OpenStream ¶
OpenStream is used to create a new stream
func (*Session) RemoteAddr ¶
RemoteAddr is used to get the address of remote end of the underlying connection
type Stream ¶
type Stream struct {
// contains filtered or unexported fields
}
Stream is used to represent a logical stream within a session. Methods on Stream are safe to call concurrently with one another, but all Read calls must be on the same goroutine and all Write calls must be on the same goroutine.
func (*Stream) Read ¶
Read is used to read from the stream. It is safe to call Write, Read, and/or Close concurrently with each other, but calls to Read are not reentrant and should not be called from multiple goroutines. Multiple Read goroutines would receive different chunks of data from the Stream and be unable to reassemble them in order or along message boundaries, and may encounter deadlocks.
func (*Stream) RemoteAddr ¶
RemoteAddr returns the remote address
func (*Stream) SetDeadline ¶
SetDeadline sets the read and write deadlines
func (*Stream) SetReadDeadline ¶
SetReadDeadline sets the deadline for blocked and future Read calls.
func (*Stream) SetWriteDeadline ¶
SetWriteDeadline sets the deadline for blocked and future Write calls
Source Files