msgio

README

go-msgio - Message IO

This is a simple package that helps read and write length-delimited slices. It's helpful for building wire protocols.

Usage

Reading

import "github.com/jbenet/msgio"
rdr := ... // some reader from a wire
mrdr := msgio.NewReader(rdr)

for {
  msg, err := mrdr.ReadMsg()
  if err != nil {
    return err
  }

  doSomething(msg)
}

Writing

import "github.com/jbenet/msgio"
wtr := genReader()
mwtr := msgio.NewWriter(wtr)

for {
  msg := genMessage()
  err := mwtr.WriteMsg(msg)
  if err != nil {
    return err
  }
}

Duplex

import "github.com/jbenet/msgio"
rw := genReadWriter()
mrw := msgio.NewReadWriter(rw)

for {
  msg, err := mrdr.ReadMsg()
  if err != nil {
    return err
  }

  // echo it back :)
  err = mwtr.WriteMsg(msg)
  if err != nil {
    return err
  }
}

Channels

import "github.com/jbenet/msgio"
rw := genReadWriter()
rch := msgio.NewReadChannel(rw)
wch := msgio.NewWriteChannel(rw)

for {
  msg, err := <-rch
  if err != nil {
    return err
  }

  // echo it back :)
  wch<- rw
}

Documentation

Index

• Variables
• func LimitedReader(r io.Reader) (io.Reader, error)
• func ReadLen(r io.Reader, buf []byte) (int, error)
• func WriteLen(w io.Writer, l int) error
• type Chan
  • func NewChan(chanSize int) *Chan
  • func (s *Chan) Close()
  • func (s *Chan) ReadFrom(r io.Reader)
  • func (s *Chan) ReadFromWithPool(r io.Reader, p *mpool.Pool)
  • func (s *Chan) WriteTo(w io.Writer)
• type LimitedWriter
  • func NewLimitedWriter(w io.Writer) *LimitedWriter
  • func (w *LimitedWriter) Flush() error
  • func (w *LimitedWriter) Write(buf []byte) (n int, err error)
• type ReadCloser
  • func NewReader(r io.Reader) ReadCloser
  • func NewReaderWithPool(r io.Reader, p *mpool.Pool) ReadCloser
  • func NewVarintReader(r io.Reader) ReadCloser
  • func NewVarintReaderWithPool(r io.Reader, p *mpool.Pool) ReadCloser
• type ReadWriteCloser
  • func Combine(w Writer, r Reader) ReadWriteCloser
  • func NewReadWriter(rw io.ReadWriter) ReadWriteCloser
• type ReadWriter
• type Reader
• type WriteCloser
  • func NewVarintWriter(w io.Writer) WriteCloser
  • func NewWriter(w io.Writer) WriteCloser
• type Writer

Variables

var ErrMsgTooLarge = errors.New("message too large")

ErrMsgTooLarge is returned when the message length is exessive

var NBO = binary.BigEndian

NBO is NetworkByteOrder

Functions

func LimitedReader

func LimitedReader(r io.Reader) (io.Reader, error)

LimitedReader wraps an io.Reader with a msgio framed reader. The LimitedReader will return a reader which will io.EOF when the msg length is done.

func ReadLen

func ReadLen(r io.Reader, buf []byte) (int, error)

ReadLen reads a length from the given reader. if buf is non-nil, it reuses the buffer. Ex:

l, err := ReadLen(r, nil)
_, err := ReadLen(r, buf)

func WriteLen

func WriteLen(w io.Writer, l int) error

WriteLen writes a length to the given writer.

Types

type Chan

type Chan struct {
	MsgChan   chan []byte
	ErrChan   chan error
	CloseChan chan bool
}

Chan is a msgio duplex channel. It is used to have a channel interface around a msgio.Reader or Writer.

func NewChan

func NewChan(chanSize int) *Chan

NewChan constructs a Chan with a given buffer size.

func (*Chan) Close

func (s *Chan) Close()

Close the Chan

func (*Chan) ReadFrom

func (s *Chan) ReadFrom(r io.Reader)

ReadFrom wraps the given io.Reader with a msgio.Reader, reads all messages, ands sends them down the channel.

func (*Chan) ReadFromWithPool

func (s *Chan) ReadFromWithPool(r io.Reader, p *mpool.Pool)

ReadFromWithPool wraps the given io.Reader with a msgio.Reader, reads all messages, ands sends them down the channel. Uses given Pool

func (*Chan) WriteTo

func (s *Chan) WriteTo(w io.Writer)

WriteTo wraps the given io.Writer with a msgio.Writer, listens on the channel and writes all messages to the writer.

type LimitedWriter

type LimitedWriter struct {
	W io.Writer
	B bytes.Buffer
	M sync.Mutex
}

func NewLimitedWriter

func NewLimitedWriter(w io.Writer) *LimitedWriter

LimitedWriter wraps an io.Writer with a msgio framed writer. It is the inverse of LimitedReader: it will buffer all writes until "Flush" is called. When Flush is called, it will write the size of the buffer first, flush the buffer, reset the buffer, and begin accept more incoming writes.

func (*LimitedWriter) Flush

func (w *LimitedWriter) Flush() error

func (*LimitedWriter) Write

func (w *LimitedWriter) Write(buf []byte) (n int, err error)

type ReadCloser

type ReadCloser interface {
	Reader
	io.Closer
}

ReadCloser combines a Reader and Closer.

func NewReader

func NewReader(r io.Reader) ReadCloser

NewReader wraps an io.Reader with a msgio framed reader. The msgio.Reader will read whole messages at a time (using the length). Assumes an equivalent writer on the other side.

func NewReaderWithPool

func NewReaderWithPool(r io.Reader, p *mpool.Pool) ReadCloser

NewReaderWithPool wraps an io.Reader with a msgio framed reader. The msgio.Reader will read whole messages at a time (using the length). Assumes an equivalent writer on the other side. It uses a given mpool.Pool

func NewVarintReader

func NewVarintReader(r io.Reader) ReadCloser

NewVarintReader wraps an io.Reader with a varint msgio framed reader. The msgio.Reader will read whole messages at a time (using the length). Varints read according to https://golang.org/pkg/encoding/binary/#ReadUvarint Assumes an equivalent writer on the other side.

func NewVarintReaderWithPool

func NewVarintReaderWithPool(r io.Reader, p *mpool.Pool) ReadCloser

NewVarintReaderWithPool wraps an io.Reader with a varint msgio framed reader. The msgio.Reader will read whole messages at a time (using the length). Varints read according to https://golang.org/pkg/encoding/binary/#ReadUvarint Assumes an equivalent writer on the other side. It uses a given mpool.Pool

type ReadWriteCloser

type ReadWriteCloser interface {
	Reader
	Writer
	io.Closer
}

ReadWriteCloser combines a Reader, a Writer, and Closer.

func Combine

func Combine(w Writer, r Reader) ReadWriteCloser

Combine wraps a pair of msgio.Writer and msgio.Reader with a msgio.ReadWriter.

func NewReadWriter

func NewReadWriter(rw io.ReadWriter) ReadWriteCloser

NewReadWriter wraps an io.ReadWriter with a msgio.ReadWriter. Writing and Reading will be appropriately framed.

type ReadWriter

type ReadWriter interface {
	Reader
	Writer
}

ReadWriter combines a Reader and Writer.

type Reader

type Reader interface {

	// Read reads the next message from the Reader.
	// The client must pass a buffer large enough, or io.ErrShortBuffer will be
	// returned.
	Read([]byte) (int, error)

	// ReadMsg reads the next message from the Reader.
	// Uses a mpool.Pool internally to reuse buffers. io.ErrShortBuffer will
	// be returned if the Pool.Get(...) returns nil.
	// User may call ReleaseMsg(msg) to signal a buffer can be reused.
	ReadMsg() ([]byte, error)

	// ReleaseMsg signals a buffer can be reused.
	ReleaseMsg([]byte)

	// NextMsgLen returns the length of the next (peeked) message. Does
	// not destroy the message or have other adverse effects
	NextMsgLen() (int, error)
}

Reader is the msgio Reader interface. It reads len-framed messages.

type WriteCloser

type WriteCloser interface {
	Writer
	io.Closer
}

WriteCloser is a Writer + Closer interface. Like in `golang/pkg/io`

func NewVarintWriter

func NewVarintWriter(w io.Writer) WriteCloser

NewVarintWriter wraps an io.Writer with a varint msgio framed writer. The msgio.Writer will write the length prefix of every message written as a varint, using https://golang.org/pkg/encoding/binary/#PutUvarint

func NewWriter

func NewWriter(w io.Writer) WriteCloser

NewWriter wraps an io.Writer with a msgio framed writer. The msgio.Writer will write the length prefix of every message written.

type Writer

type Writer interface {

	// Write writes passed in buffer as a single message.
	Write([]byte) (int, error)

	// WriteMsg writes the msg in the passed in buffer.
	WriteMsg([]byte) error
}

Writer is the msgio Writer interface. It writes len-framed messages.