multiplex

README

go-mplex

A super simple stream muxing library implementing mplex.

Usage

mplex := multiplex.NewMultiplex(mysocket)

s, _ := mplex.NewStream()
s.Write([]byte("Hello World!"))
s.Close()

os, _ := mplex.Accept()
// echo back everything received
io.Copy(os, os)

---

The last gx published version of this module was: 0.2.35: QmWGQQ6Tz8AdUpxktLf3zgnVN9Vy8fcWVezZJSU3ZmiANj

Documentation

Index

• Variables
• type Multiplex
  • func NewMultiplex(con net.Conn, initiator bool) *Multiplex
  • func (m *Multiplex) Accept() (*Stream, error)
  • func (mp *Multiplex) Close() error
  • func (mp *Multiplex) IsClosed() bool
  • func (mp *Multiplex) NewNamedStream(ctx context.Context, name string) (*Stream, error)
  • func (mp *Multiplex) NewStream(ctx context.Context) (*Stream, error)
• type Stream
  • func (s *Stream) Close() error
  • func (s *Stream) CloseRead() error
  • func (s *Stream) CloseWrite() error
  • func (s *Stream) Name() string
  • func (s *Stream) Read(b []byte) (int, error)
  • func (s *Stream) Reset() error
  • 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) Write(b []byte) (int, error)

Variables

var (
	ResetStreamTimeout = 2 * time.Minute

	WriteCoalesceDelay = 100 * time.Microsecond
)

var (
	ErrStreamReset  = errors.New("stream reset")
	ErrStreamClosed = errors.New("closed stream")
)

var ErrInvalidState = errors.New("received an unexpected message from the peer")

ErrInvalidState is returned when the other side does something it shouldn't. In this case, we close the connection to be safe.

var ErrShutdown = errors.New("session shut down")

ErrShutdown is returned when operating on a shutdown session

var ErrTwoInitiators = errors.New("two initiators")

ErrTwoInitiators is returned when both sides think they're the initiator

var MaxMessageSize = 1 << 20

var ReceiveTimeout = 5 * time.Second

Max time to block waiting for a slow reader to read from a stream before resetting it. Preferably, we'd have some form of back-pressure mechanism but we don't have that in this protocol.

Types

type Multiplex

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

Multiplex is a mplex session.

func NewMultiplex

func NewMultiplex(con net.Conn, initiator bool) *Multiplex

NewMultiplex creates a new multiplexer session.

func (*Multiplex) Accept

func (m *Multiplex) Accept() (*Stream, error)

Accept accepts the next stream from the connection.

func (*Multiplex) Close

func (mp *Multiplex) Close() error

Close closes the session.

func (*Multiplex) IsClosed

func (mp *Multiplex) IsClosed() bool

IsClosed returns true if the session is closed.

func (*Multiplex) NewNamedStream

func (mp *Multiplex) NewNamedStream(ctx context.Context, name string) (*Stream, error)

NewNamedStream creates a new named stream.

func (*Multiplex) NewStream

func (mp *Multiplex) NewStream(ctx context.Context) (*Stream, error)

NewStream creates a new stream.

type Stream

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

func (*Stream) Close

func (s *Stream) Close() error

func (*Stream) CloseRead

func (s *Stream) CloseRead() error

func (*Stream) CloseWrite

func (s *Stream) CloseWrite() error

func (*Stream) Name

func (s *Stream) Name() string

func (*Stream) Read

func (s *Stream) Read(b []byte) (int, error)

func (*Stream) Reset

func (s *Stream) Reset() error

func (*Stream) SetDeadline

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

func (*Stream) SetReadDeadline

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

func (*Stream) SetWriteDeadline

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

func (*Stream) Write

func (s *Stream) Write(b []byte) (int, error)