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
- 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 the associated Godoc.
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)
}
The last gx published version of this module was: 1.1.5: QmUNMbRUsVYHi1D14annF7Rr7pQAX7TNLwpRCa975ojKnw
Documentation
¶
Index ¶
- Variables
- func VerifyConfig(config *Config) error
- type Config
- type Error
- type GoAwayError
- type MemoryManager
- type Session
- func (s *Session) Accept() (net.Conn, error)
- func (s *Session) AcceptStream() (*Stream, error)
- func (s *Session) Addr() net.Addr
- func (s *Session) Close() error
- func (s *Session) CloseChan() <-chan struct{}
- func (s *Session) CloseWithError(errCode uint32) error
- 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(ctx context.Context) (net.Conn, error)
- func (s *Session) OpenStream(ctx context.Context) (*Stream, error)
- func (s *Session) Ping() (dur time.Duration, err error)
- func (s *Session) RTT() time.Duration
- func (s *Session) RemoteAddr() net.Addr
- type Stream
- func (s *Stream) Close() error
- func (s *Stream) CloseRead() error
- func (s *Stream) CloseWrite() 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) Reset() error
- func (s *Stream) ResetWithError(errCode uint32) error
- 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) StreamID() uint32
- func (s *Stream) Write(b []byte) (int, error)
- type StreamError
Constants ¶
This section is empty.
Variables ¶
var ( // ErrInvalidVersion means we received a frame with an // invalid version ErrInvalidVersion = &Error{msg: "invalid protocol version"} // ErrInvalidMsgType means we received a frame with an // invalid message type ErrInvalidMsgType = &Error{msg: "invalid msg type"} // ErrSessionShutdown is used if there is a shutdown during // an operation ErrSessionShutdown = &GoAwayError{ErrorCode: goAwayNormal, Remote: false} // ErrStreamsExhausted is returned if we have no more // stream ids to issue ErrStreamsExhausted = &Error{msg: "streams exhausted"} // ErrDuplicateStream is used if a duplicate stream is // opened inbound ErrDuplicateStream = &Error{msg: "duplicate stream initiated"} // ErrReceiveWindowExceeded indicates the window was exceeded ErrRecvWindowExceeded = &Error{msg: "recv window exceeded"} // ErrTimeout is used when we reach an IO deadline ErrTimeout = &Error{msg: "i/o deadline reached", timeout: true, temporary: true} // ErrStreamClosed is returned when using a closed stream ErrStreamClosed = &Error{msg: "stream closed"} // ErrUnexpectedFlag is set when we get an unexpected flag ErrUnexpectedFlag = &Error{msg: "unexpected flag"} // ErrRemoteGoAway is used when we get a go away from the other side with error code // goAwayNormal(0). ErrRemoteGoAway = &GoAwayError{Remote: true, ErrorCode: goAwayNormal} // ErrStreamReset is sent if a stream is reset. This can happen // if the backlog is exceeded, or if there was a remote GoAway. ErrStreamReset = &Error{msg: "stream reset"} // ErrConnectionWriteTimeout indicates that we hit the "safety valve" // timeout writing to the underlying stream connection. ErrConnectionWriteTimeout = &Error{msg: "connection write timeout", timeout: true} // ErrKeepAliveTimeout is sent if a missed keepalive caused the stream close ErrKeepAliveTimeout = &Error{msg: "keepalive timeout", timeout: true} )
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
// PingBacklog is used to limit how many ping acks we can queue.
PingBacklog 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
// MeasureRTTInterval is how often to re-measure the round trip time
MeasureRTTInterval 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
// MaxIncomingStreams is maximum number of concurrent incoming streams
// that we accept. If the peer tries to open more streams, those will be
// reset immediately.
MaxIncomingStreams uint32
// InitialStreamWindowSize is used to control the initial
// window size that we allow for a stream.
InitialStreamWindowSize uint32
// MaxStreamWindowSize is used to control the maximum
// window size that we allow for a stream.
MaxStreamWindowSize uint32
// LogOutput is used to control the log destination
LogOutput io.Writer
// ReadBufSize controls the size of the read buffer.
//
// Set to 0 to disable it.
ReadBufSize int
// WriteCoalesceDelay is the maximum amount of time we'll delay
// coalescing a packet before sending it. This should be on the order of
// micro-milliseconds.
WriteCoalesceDelay time.Duration
// MaxMessageSize is the maximum size of a message that we'll send on a
// stream. This ensures that a single stream doesn't hog a connection.
MaxMessageSize uint32
}
Config is used to tune the Yamux session
func DefaultConfig ¶
func DefaultConfig() *Config
DefaultConfig is used to return a default configuration
type GoAwayError ¶
func (*GoAwayError) Error ¶
func (e *GoAwayError) Error() string
func (*GoAwayError) Is ¶
func (e *GoAwayError) Is(target error) bool
func (*GoAwayError) Temporary ¶
func (e *GoAwayError) Temporary() bool
func (*GoAwayError) Timeout ¶
func (e *GoAwayError) Timeout() bool
type MemoryManager ¶
type MemoryManager interface {
ReserveMemory(size int, prio uint8) error
// ReleaseMemory explicitly releases memory previously reserved with ReserveMemory
ReleaseMemory(size int)
// Done ends the span and releases associated resources.
Done()
}
The MemoryManager allows management of memory allocations. Memory is allocated: 1. When opening / accepting a new stream. This uses the highest priority. 2. When trying to increase the stream receive window. This uses a lower priority. This is a subset of the libp2p's resource manager ResourceScopeSpan interface.
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 ¶
Client is used to initialize a new client-side connection. There must be at most one client-side connection.
func Server ¶
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) Close ¶
Close is used to close the session and all streams. It doesn't 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) CloseWithError ¶
CloseWithError is used to close the session and all streams after sending a GoAway message with errCode. Blocks for ConnectionWriteTimeout to write the GoAway message.
The GoAway may not actually be sent depending on the semantics of the underlying net.Conn. For TCP connections, it may be dropped depending on LINGER value or if there's unread data in the kernel receive buffer.
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.
func (*Stream) CloseRead ¶
CloseRead is used to close the stream for reading. Note: Remote is not notified.
func (*Stream) CloseWrite ¶
CloseWrite is used to close the stream for writing.
func (*Stream) RemoteAddr ¶
RemoteAddr returns the remote address
func (*Stream) ResetWithError ¶
func (*Stream) SetDeadline ¶
SetDeadline sets the read and write deadlines
func (*Stream) SetReadDeadline ¶
SetReadDeadline sets the deadline for future Read calls.
func (*Stream) SetWriteDeadline ¶
SetWriteDeadline sets the deadline for future Write calls
type StreamError ¶
A StreamError is used for errors returned from Read and Write calls after the stream is Reset
func (*StreamError) Error ¶
func (s *StreamError) Error() string
func (*StreamError) Is ¶
func (s *StreamError) Is(target error) bool
Source Files