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)

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(name string) (*Stream, error)
  • func (mp *Multiplex) NewStream() (*Stream, error)
• type Stream
  • func (s *Stream) Close() 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 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
}

func NewMultiplex

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

func (*Multiplex) Accept

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

func (*Multiplex) Close

func (mp *Multiplex) Close() error

func (*Multiplex) IsClosed

func (mp *Multiplex) IsClosed() bool

func (*Multiplex) NewNamedStream

func (mp *Multiplex) NewNamedStream(name string) (*Stream, error)

func (*Multiplex) NewStream

func (mp *Multiplex) NewStream() (*Stream, error)

type Stream

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

func (*Stream) Close

func (s *Stream) Close() 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)